core, network: add Notary module

pkg/config/notary_config.go

@@ -0,0 +1,7 @@
+package config
+
+// P2PNotary stores configuration for Notary node service.
+type P2PNotary struct {
+	Enabled      bool   `yaml:"Enabled"`
+	UnlockWallet Wallet `yaml:"UnlockWallet"`
+}

pkg/config/protocol_config.go

@@ -20,6 +20,8 @@ type (
 		RemoveUntraceableBlocks bool `yaml:"RemoveUntraceableBlocks"`
 		// MaxTraceableBlocks is the length of the chain accessible to smart contracts.
 		MaxTraceableBlocks uint32 `yaml:"MaxTraceableBlocks"`
+		// P2PNotary stores configuration for P2P notary node service
+		P2PNotary P2PNotary `yaml:"P2PNotary"`
 		// P2PSigExtensions enables additional signature-related logic.
 		P2PSigExtensions bool `yaml:"P2PSigExtensions"`
 		// ReservedAttributes allows to have reserved attributes range for experimental or private purposes.

pkg/core/blockchain.go

@@ -122,6 +122,8 @@ type Blockchain struct {
 	// postBlock is a set of callback methods which should be run under the Blockchain lock after new block is persisted.
 	// Block's transactions are passed via mempool.
 	postBlock []func(blockchainer.Blockchainer, *mempool.Pool, *block.Block)
+	// poolTxWithDataCallbacks is a set of callback methods which should be run nuder the Blockchain lock after successful PoolTxWithData invocation.
+	poolTxWithDataCallbacks []func(t *transaction.Transaction, data interface{})
 
 	sbCommittee keys.PublicKeys
 
@@ -201,6 +203,12 @@ func (bc *Blockchain) SetOracle(mod services.Oracle) {
 	bc.contracts.Designate.OracleService.Store(mod)
 }
 
+// SetNotary sets notary module. It doesn't protected by mutex and
+// must be called before `bc.Run()` to avoid data race.
+func (bc *Blockchain) SetNotary(mod services.Notary) {
+	bc.contracts.Designate.NotaryService.Store(mod)
+}
+
 func (bc *Blockchain) init() error {
 	// If we could not find the version in the Store, we know that there is nothing stored.
 	ver, err := bc.dao.GetVersion()
@@ -1671,7 +1679,19 @@ func (bc *Blockchain) PoolTxWithData(t *transaction.Transaction, data interface{
 			return err
 		}
 	}
-	return bc.verifyAndPoolTx(t, mp, feer, data)
+	if err := bc.verifyAndPoolTx(t, mp, feer, data); err != nil {
+		return err
+	}
+	for _, f := range bc.poolTxWithDataCallbacks {
+		f(t, data)
+	}
+	return nil
+}
+
+// RegisterPoolTxWithDataCallback registers new callback function which is called
+// under the Blockchain lock after successful PoolTxWithData invocation.
+func (bc *Blockchain) RegisterPoolTxWithDataCallback(f func(t *transaction.Transaction, data interface{})) {
+	bc.poolTxWithDataCallbacks = append(bc.poolTxWithDataCallbacks, f)
 }
 
 //GetStandByValidators returns validators from the configuration.
@@ -1892,6 +1912,11 @@ func (bc *Blockchain) newInteropContext(trigger trigger.Type, d dao.DAO, block *
 	return ic
 }
 
+// P2PNotaryModuleEnabled defines whether P2P notary module is enabled.
+func (bc *Blockchain) P2PNotaryModuleEnabled() bool {
+	return bc.config.P2PNotary.Enabled
+}
+
 // P2PSigExtensionsEnabled defines whether P2P signature extensions are enabled.
 func (bc *Blockchain) P2PSigExtensionsEnabled() bool {
 	return bc.config.P2PSigExtensions

pkg/core/blockchainer/blockchainer.go

@@ -64,7 +64,9 @@ type Blockchainer interface {
 	ManagementContractHash() util.Uint160
 	PoolTx(t *transaction.Transaction, pools ...*mempool.Pool) error
 	PoolTxWithData(t *transaction.Transaction, data interface{}, mp *mempool.Pool, feer mempool.Feer, verificationFunction func(bc Blockchainer, t *transaction.Transaction, data interface{}) error) error
+	RegisterPoolTxWithDataCallback(f func(t *transaction.Transaction, data interface{}))
 	RegisterPostBlock(f func(Blockchainer, *mempool.Pool, *block.Block))
+	SetNotary(mod services.Notary)
 	SubscribeForBlocks(ch chan<- *block.Block)
 	SubscribeForExecutions(ch chan<- *state.AppExecResult)
 	SubscribeForNotifications(ch chan<- *state.NotificationEvent)

pkg/core/blockchainer/services/notary.go

@@ -0,0 +1,8 @@
+package services
+
+import "github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+
+// Notary is a Notary module interface.
+type Notary interface {
+	UpdateNotaryNodes(pubs keys.PublicKeys)
+}

pkg/core/helper_test.go

@@ -531,6 +531,14 @@ func invokeContractMethodBy(t *testing.T, chain *Blockchain, signer *wallet.Acco
 	return &res[0], nil
 }
 
+func transferTokenFromMultisigAccountCheckOK(t *testing.T, chain *Blockchain, to, tokenHash util.Uint160, amount int64, additionalArgs ...interface{}) {
+	transferTx := transferTokenFromMultisigAccount(t, chain, to, tokenHash, amount, additionalArgs...)
+	res, err := chain.GetAppExecResults(transferTx.Hash(), trigger.Application)
+	require.NoError(t, err)
+	require.Equal(t, vm.HaltState, res[0].VMState)
+	require.Equal(t, 0, len(res[0].Stack))
+}
+
 func transferTokenFromMultisigAccount(t *testing.T, chain *Blockchain, to, tokenHash util.Uint160, amount int64, additionalArgs ...interface{}) *transaction.Transaction {
 	transferTx := newNEP17Transfer(tokenHash, testchain.MultisigScriptHash(), to, amount, additionalArgs...)
 	transferTx.SystemFee = 100000000

pkg/core/mempool/feer.go

@@ -12,4 +12,5 @@ type Feer interface {
 	GetUtilityTokenBalance(util.Uint160) *big.Int
 	BlockHeight() uint32
 	P2PSigExtensionsEnabled() bool
+	P2PNotaryModuleEnabled() bool
 }

pkg/core/mempool/mem_pool.go

@@ -69,6 +69,8 @@ type Pool struct {
 
 	resendThreshold uint32
 	resendFunc      func(*transaction.Transaction, interface{})
+	// removeStaleCallback is a callback method which is called after item is removed from the mempool.
+	removeStaleCallback func(*transaction.Transaction, interface{})
 }
 
 func (p items) Len() int           { return len(p) }
@@ -325,7 +327,10 @@ func (mp *Pool) RemoveStale(isOK func(*transaction.Transaction) bool, feer Feer)
 		mp.conflicts = make(map[util.Uint256][]util.Uint256)
 	}
 	height := feer.BlockHeight()
-	var staleItems []item
+	var (
+		staleItems   []item
+		removedItems []item
+	)
 	for _, itm := range mp.verifiedTxes {
 		if isOK(itm.txn) && mp.checkPolicy(itm.txn, policyChanged) && mp.tryAddSendersFee(itm.txn, feer, true) {
 			newVerifiedTxes = append(newVerifiedTxes, itm)
@@ -348,11 +353,17 @@ func (mp *Pool) RemoveStale(isOK func(*transaction.Transaction) bool, feer Feer)
 			if attrs := itm.txn.GetAttributes(transaction.OracleResponseT); len(attrs) != 0 {
 				delete(mp.oracleResp, attrs[0].Value.(*transaction.OracleResponse).ID)
 			}
+			if feer.P2PSigExtensionsEnabled() && feer.P2PNotaryModuleEnabled() && mp.removeStaleCallback != nil {
+				removedItems = append(removedItems, itm)
+			}
 		}
 	}
 	if len(staleItems) != 0 {
 		go mp.resendStaleItems(staleItems)
 	}
+	if len(removedItems) != 0 {
+		go mp.postRemoveStale(removedItems)
+	}
 	mp.verifiedTxes = newVerifiedTxes
 	mp.lock.Unlock()
 }
@@ -398,12 +409,25 @@ func (mp *Pool) SetResendThreshold(h uint32, f func(*transaction.Transaction, in
 	mp.resendFunc = f
 }
 
+// SetRemoveStaleCallback registers new callback method which should be called after mempool item is kicked off.
+func (mp *Pool) SetRemoveStaleCallback(f func(t *transaction.Transaction, data interface{})) {
+	mp.lock.Lock()
+	defer mp.lock.Unlock()
+	mp.removeStaleCallback = f
+}
+
 func (mp *Pool) resendStaleItems(items []item) {
 	for i := range items {
 		mp.resendFunc(items[i].txn, items[i].data)
 	}
 }
 
+func (mp *Pool) postRemoveStale(items []item) {
+	for i := range items {
+		mp.removeStaleCallback(items[i].txn, items[i].data)
+	}
+}
+
 // TryGetValue returns a transaction and its fee if it exists in the memory pool.
 func (mp *Pool) TryGetValue(hash util.Uint256) (*transaction.Transaction, bool) {
 	mp.lock.RLock()

pkg/core/mempool/mem_pool_test.go

@@ -44,6 +44,10 @@ func (fs *FeerStub) P2PSigExtensionsEnabled() bool {
 	return fs.p2pSigExt
 }
 
+func (fs *FeerStub) P2PNotaryModuleEnabled() bool {
+	return false
+}
+
 func testMemPoolAddRemoveWithFeer(t *testing.T, fs Feer) {
 	mp := New(10, 0)
 	tx := transaction.New(netmode.UnitTestNet, []byte{byte(opcode.PUSH1)}, 0)

pkg/core/native/designate.go

@@ -37,6 +37,8 @@ type Designate struct {
 	p2pSigExtensionsEnabled bool
 
 	OracleService atomic.Value
+	// NotaryService represents Notary node module.
+	NotaryService atomic.Value
 }
 
 type roleData struct {
@@ -183,10 +185,15 @@ func (s *Designate) updateCachedRoleData(v *atomic.Value, d dao.DAO, r Role) err
 		addr:   s.hashFromNodes(r, nodeKeys),
 		height: height,
 	})
-	if r == RoleOracle {
+	switch r {
+	case RoleOracle:
 		if orc, _ := s.OracleService.Load().(services.Oracle); orc != nil {
 			orc.UpdateOracleNodes(nodeKeys.Copy())
 		}
+	case RoleP2PNotary:
+		if ntr, _ := s.NotaryService.Load().(services.Notary); ntr != nil {
+			ntr.UpdateNotaryNodes(nodeKeys.Copy())
+		}
 	}
 	return nil
 }

pkg/network/helper_test.go

@@ -143,6 +143,10 @@ func (chain *testChain) P2PSigExtensionsEnabled() bool {
 	return true
 }
 
+func (chain *testChain) P2PNotaryModuleEnabled() bool {
+	return false
+}
+
 func (chain *testChain) GetMaxBlockSystemFee() int64 {
 	panic("TODO")
 }
@@ -285,6 +289,12 @@ func (chain *testChain) PoolTx(tx *transaction.Transaction, _ ...*mempool.Pool)
 func (chain testChain) SetOracle(services.Oracle) {
 	panic("TODO")
 }
+func (chain *testChain) RegisterPoolTxWithDataCallback(f func(t *transaction.Transaction, data interface{})) {
+	panic("TODO")
+}
+func (chain *testChain) SetNotary(notary services.Notary) {
+	panic("TODO")
+}
 func (chain *testChain) SubscribeForBlocks(ch chan<- *block.Block) {
 	chain.blocksCh = append(chain.blocksCh, ch)
 }

pkg/network/notary_feer.go

@@ -32,6 +32,11 @@ func (f NotaryFeer) P2PSigExtensionsEnabled() bool {
 	return f.bc.P2PSigExtensionsEnabled()
 }
 
+// P2PNotaryModuleEnabled implements mempool.Feer interface.
+func (f NotaryFeer) P2PNotaryModuleEnabled() bool {
+	return f.bc.P2PNotaryModuleEnabled()
+}
+
 // NewNotaryFeer returns new NotaryFeer instance.
 func NewNotaryFeer(bc blockchainer.Blockchainer) NotaryFeer {
 	return NotaryFeer{

pkg/network/server.go

@@ -21,6 +21,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/network/capability"
 	"github.com/nspcc-dev/neo-go/pkg/network/extpool"
 	"github.com/nspcc-dev/neo-go/pkg/network/payload"
+	"github.com/nspcc-dev/neo-go/pkg/services/notary"
 	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"go.uber.org/atomic"
@@ -69,7 +70,8 @@ type (
 		consensus         consensus.Service
 		notaryRequestPool *mempool.Pool
 		extensiblePool    *extpool.Pool
-		NotaryFeer        NotaryFeer
+		notaryFeer        NotaryFeer
+		notaryModule      *notary.Notary
 
 		lock  sync.RWMutex
 		peers map[Peer]bool
@@ -134,13 +136,37 @@ func newServerFromConstructors(config ServerConfig, chain blockchainer.Blockchai
 		transactions:      make(chan *transaction.Transaction, 64),
 	}
 	if chain.P2PSigExtensionsEnabled() {
-		s.NotaryFeer = NewNotaryFeer(chain)
+		s.notaryFeer = NewNotaryFeer(chain)
 		s.notaryRequestPool = mempool.New(chain.GetConfig().P2PNotaryRequestPayloadPoolSize, 1)
 		chain.RegisterPostBlock(func(bc blockchainer.Blockchainer, txpool *mempool.Pool, _ *block.Block) {
 			s.notaryRequestPool.RemoveStale(func(t *transaction.Transaction) bool {
 				return bc.IsTxStillRelevant(t, txpool, true)
-			}, s.NotaryFeer)
+			}, s.notaryFeer)
 		})
+		if chain.GetConfig().P2PNotary.Enabled {
+			n, err := notary.NewNotary(chain, s.log, func(tx *transaction.Transaction) error {
+				r := s.RelayTxn(tx)
+				if r != RelaySucceed {
+					return fmt.Errorf("can't pool notary tx: hash %s, reason: %d", tx.Hash().StringLE(), byte(r))
+				}
+				return nil
+			})
+			if err != nil {
+				return nil, fmt.Errorf("failed to create Notary module: %w", err)
+			}
+			s.notaryModule = n
+			chain.SetNotary(n)
+			chain.RegisterPoolTxWithDataCallback(func(_ *transaction.Transaction, data interface{}) {
+				if notaryRequest, ok := data.(*payload.P2PNotaryRequest); ok {
+					s.notaryModule.OnNewRequest(notaryRequest)
+				}
+			})
+			chain.RegisterPostBlock(func(bc blockchainer.Blockchainer, pool *mempool.Pool, b *block.Block) {
+				s.notaryModule.PostPersist(bc, pool, b)
+			})
+		}
+	} else if chain.GetConfig().P2PNotary.Enabled {
+		return nil, errors.New("P2PSigExtensions are disabled, but Notary service is enable")
 	}
 	s.bQueue = newBlockQueue(maxBlockBatch, chain, log, func(b *block.Block) {
 		if !s.consensusStarted.Load() {
@@ -805,7 +831,7 @@ func (s *Server) handleP2PNotaryRequestCmd(r *payload.P2PNotaryRequest) error {
 
 // verifyAndPoolNotaryRequest verifies NotaryRequest payload and adds it to the payload mempool.
 func (s *Server) verifyAndPoolNotaryRequest(r *payload.P2PNotaryRequest) RelayReason {
-	if err := s.chain.PoolTxWithData(r.FallbackTransaction, r, s.notaryRequestPool, s.NotaryFeer, verifyNotaryRequest); err != nil {
+	if err := s.chain.PoolTxWithData(r.FallbackTransaction, r, s.notaryRequestPool, s.notaryFeer, verifyNotaryRequest); err != nil {
 		switch {
 		case errors.Is(err, core.ErrAlreadyExists):
 			return RelayAlreadyExists
@@ -827,7 +853,8 @@ func verifyNotaryRequest(bc blockchainer.Blockchainer, _ *transaction.Transactio
 	if err := bc.VerifyWitness(payer, r, &r.Witness, bc.GetPolicer().GetMaxVerificationGAS()); err != nil {
 		return fmt.Errorf("bad P2PNotaryRequest payload witness: %w", err)
 	}
-	if r.FallbackTransaction.Sender() != bc.GetNotaryContractScriptHash() {
+	notaryHash := bc.GetNotaryContractScriptHash()
+	if r.FallbackTransaction.Sender() != notaryHash {
 		return errors.New("P2PNotary contract should be a sender of the fallback transaction")
 	}
 	depositExpiration := bc.GetNotaryDepositExpiration(payer)
@@ -1168,6 +1195,13 @@ func (s *Server) initStaleMemPools() {
 	mp.SetResendThreshold(uint32(threshold), s.broadcastTX)
 	if s.chain.P2PSigExtensionsEnabled() {
 		s.notaryRequestPool.SetResendThreshold(uint32(threshold), s.broadcastP2PNotaryRequestPayload)
+		if s.chain.GetConfig().P2PNotary.Enabled {
+			s.notaryRequestPool.SetRemoveStaleCallback(func(_ *transaction.Transaction, data interface{}) {
+				if notaryRequest, ok := data.(*payload.P2PNotaryRequest); ok {
+					s.notaryModule.OnRequestRemoval(notaryRequest)
+				}
+			})
+		}
 	}
 }
 

pkg/network/server_test.go

@@ -850,6 +850,7 @@ func (f feerStub) FeePerByte() int64                            { return 1 }
 func (f feerStub) GetUtilityTokenBalance(util.Uint160) *big.Int { return big.NewInt(100000000) }
 func (f feerStub) BlockHeight() uint32                          { return f.blockHeight }
 func (f feerStub) P2PSigExtensionsEnabled() bool                { return false }
+func (f feerStub) P2PNotaryModuleEnabled() bool                 { return false }
 func (f feerStub) GetBaseExecFee() int64                        { return interop.DefaultBaseExecFee }
 
 func TestMemPool(t *testing.T) {

pkg/rpc/server/server_helper_test.go

@@ -126,6 +126,10 @@ func (fs FeerStub) P2PSigExtensionsEnabled() bool {
 	return false
 }
 
+func (fs FeerStub) P2PNotaryModuleEnabled() bool {
+	return false
+}
+
 func (fs FeerStub) GetBaseExecFee() int64 {
 	return interop.DefaultBaseExecFee
 }

pkg/services/notary/node.go

@@ -0,0 +1,54 @@
+package notary
+
+import (
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
+	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/wallet"
+	"go.uber.org/zap"
+)
+
+// UpdateNotaryNodes implements Notary interface and updates current notary account.
+func (n *Notary) UpdateNotaryNodes(notaryNodes keys.PublicKeys) {
+	n.accMtx.Lock()
+	defer n.accMtx.Unlock()
+
+	if n.currAccount != nil {
+		for _, node := range notaryNodes {
+			if node.Equal(n.currAccount.PrivateKey().PublicKey()) {
+				return
+			}
+		}
+	}
+
+	var acc *wallet.Account
+	for _, node := range notaryNodes {
+		acc = n.wallet.GetAccount(node.GetScriptHash())
+		if acc != nil {
+			if acc.PrivateKey() != nil {
+				break
+			}
+			err := acc.Decrypt(n.Config.MainCfg.UnlockWallet.Password)
+			if err != nil {
+				n.Config.Log.Warn("can't unlock notary node account",
+					zap.String("address", address.Uint160ToString(acc.Contract.ScriptHash())),
+					zap.Error(err))
+				acc = nil
+			}
+			break
+		}
+	}
+
+	n.currAccount = acc
+	if acc == nil {
+		n.reqMtx.Lock()
+		n.requests = make(map[util.Uint256]*request)
+		n.reqMtx.Unlock()
+	}
+}
+
+func (n *Notary) getAccount() *wallet.Account {
+	n.accMtx.RLock()
+	defer n.accMtx.RUnlock()
+	return n.currAccount
+}

pkg/services/notary/notary.go

@@ -0,0 +1,360 @@
+package notary
+
+import (
+	"bytes"
+	"crypto/elliptic"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/nspcc-dev/neo-go/pkg/config"
+	"github.com/nspcc-dev/neo-go/pkg/core/block"
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer"
+	"github.com/nspcc-dev/neo-go/pkg/core/mempool"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/hash"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/network/payload"
+	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/vm"
+	"github.com/nspcc-dev/neo-go/pkg/vm/opcode"
+	"github.com/nspcc-dev/neo-go/pkg/wallet"
+	"go.uber.org/zap"
+)
+
+type (
+	// Notary represents Notary module.
+	Notary struct {
+		Config Config
+
+		// onTransaction is a callback for completed transactions (mains or fallbacks) sending.
+		onTransaction func(tx *transaction.Transaction) error
+
+		// reqMtx protects requests list.
+		reqMtx sync.RWMutex
+		// requests represents the map of main transactions which needs to be completed
+		// with the associated fallback transactions grouped by the main transaction hash
+		requests map[util.Uint256]*request
+
+		// accMtx protects account.
+		accMtx      sync.RWMutex
+		currAccount *wallet.Account
+		wallet      *wallet.Wallet
+	}
+
+	// Config represents external configuration for Notary module.
+	Config struct {
+		MainCfg config.P2PNotary
+		Chain   blockchainer.Blockchainer
+		Log     *zap.Logger
+	}
+)
+
+// request represents Notary service request.
+type request struct {
+	typ RequestType
+	// isSent indicates whether main transaction was successfully sent to the network.
+	isSent bool
+	main   *transaction.Transaction
+	// minNotValidBefore is the minimum NVB value among fallbacks transactions.
+	// We stop trying to send mainTx to the network if the chain reaches minNotValidBefore height.
+	minNotValidBefore uint32
+	fallbacks         []*transaction.Transaction
+	// nSigs is the number of signatures to be collected.
+	// nSigs == nKeys for standard signature request;
+	// nSigs <= nKeys for multisignature request.
+	// nSigs is 0 when all received requests were invalid, so check request.typ before access to nSigs.
+	nSigs uint8
+	// nSigsCollected is the number of already collected signatures
+	nSigsCollected uint8
+
+	// sigs is a map of partial multisig invocation scripts [opcode.PUSHDATA1+64+signatureBytes] grouped by public keys
+	sigs map[*keys.PublicKey][]byte
+}
+
+// NewNotary returns new Notary module.
+func NewNotary(bc blockchainer.Blockchainer, log *zap.Logger, onTransaction func(tx *transaction.Transaction) error) (*Notary, error) {
+	cfg := bc.GetConfig().P2PNotary
+	w := cfg.UnlockWallet
+	wallet, err := wallet.NewWalletFromFile(w.Path)
+	if err != nil {
+		return nil, err
+	}
+
+	haveAccount := false
+	for _, acc := range wallet.Accounts {
+		if err := acc.Decrypt(w.Password); err == nil {
+			haveAccount = true
+			break
+		}
+	}
+	if !haveAccount {
+		return nil, errors.New("no wallet account could be unlocked")
+	}
+
+	return &Notary{
+		requests: make(map[util.Uint256]*request),
+		Config: Config{
+			MainCfg: cfg,
+			Chain:   bc,
+			Log:     log,
+		},
+		wallet:        wallet,
+		onTransaction: onTransaction,
+	}, nil
+}
+
+// OnNewRequest is a callback method which is called after new notary request is added to the notary request pool.
+func (n *Notary) OnNewRequest(payload *payload.P2PNotaryRequest) {
+	if n.getAccount() == nil {
+		return
+	}
+
+	nvbFallback := payload.FallbackTransaction.GetAttributes(transaction.NotValidBeforeT)[0].Value.(*transaction.NotValidBefore).Height
+	nKeys := payload.MainTransaction.GetAttributes(transaction.NotaryAssistedT)[0].Value.(*transaction.NotaryAssisted).NKeys
+	typ, nSigs, pubs, validationErr := n.verifyIncompleteWitnesses(payload.MainTransaction, nKeys)
+	n.reqMtx.Lock()
+	defer n.reqMtx.Unlock()
+	r, exists := n.requests[payload.MainTransaction.Hash()]
+	if exists {
+		for _, fb := range r.fallbacks {
+			if fb.Hash().Equals(payload.FallbackTransaction.Hash()) {
+				return // then we already have processed this request
+			}
+		}
+		if nvbFallback < r.minNotValidBefore {
+			r.minNotValidBefore = nvbFallback
+		}
+		if r.typ == Unknown && validationErr == nil {
+			r.typ = typ
+			r.nSigs = nSigs
+		}
+	} else {
+		r = &request{
+			nSigs:             nSigs,
+			main:              payload.MainTransaction,
+			typ:               typ,
+			minNotValidBefore: nvbFallback,
+		}
+		n.requests[payload.MainTransaction.Hash()] = r
+	}
+	r.fallbacks = append(r.fallbacks, payload.FallbackTransaction)
+	if exists && r.typ != Unknown && r.nSigsCollected >= r.nSigs { // already collected sufficient number of signatures to complete main transaction
+		return
+	}
+	if validationErr == nil {
+	loop:
+		for i, w := range payload.MainTransaction.Scripts {
+			if payload.MainTransaction.Signers[i].Account.Equals(n.Config.Chain.GetNotaryContractScriptHash()) {
+				continue
+			}
+			if len(w.InvocationScript) != 0 && len(w.VerificationScript) != 0 {
+				switch r.typ {
+				case Signature:
+					if !exists {
+						r.nSigsCollected++
+					} else if len(r.main.Scripts[i].InvocationScript) == 0 { // need this check because signature can already be added (consider receiving the same payload multiple times)
+						r.main.Scripts[i] = w
+						r.nSigsCollected++
+					}
+					if r.nSigsCollected == r.nSigs {
+						break loop
+					}
+				case MultiSignature:
+					if r.sigs == nil {
+						r.sigs = make(map[*keys.PublicKey][]byte)
+					}
+
+					hash := r.main.GetSignedHash().BytesBE()
+					for _, pub := range pubs {
+						if r.sigs[pub] != nil {
+							continue // signature for this pub has already been added
+						}
+						if pub.Verify(w.InvocationScript[2:], hash) { // then pub is the owner of the signature
+							r.sigs[pub] = w.InvocationScript
+							r.nSigsCollected++
+							if r.nSigsCollected == r.nSigs {
+								var invScript []byte
+								for j := range pubs {
+									if sig, ok := r.sigs[pubs[j]]; ok {
+										invScript = append(invScript, sig...)
+									}
+								}
+								r.main.Scripts[i].InvocationScript = invScript
+							}
+							break loop
+						}
+					}
+					// pubKey was not found for the signature i.e. signature is bad - we're OK with that, let the fallback TX to be added
+					break loop // only one multisignature is allowed
+				}
+			}
+		}
+	}
+	if r.typ != Unknown && r.nSigsCollected == nSigs && r.minNotValidBefore > n.Config.Chain.BlockHeight() {
+		if err := n.finalize(r.main); err != nil {
+			n.Config.Log.Error("failed to finalize main transaction", zap.Error(err))
+		} else {
+			r.isSent = true
+		}
+	}
+}
+
+// OnRequestRemoval is a callback which is called after fallback transaction is removed
+// from the notary payload pool due to expiration, main tx appliance or any other reason.
+func (n *Notary) OnRequestRemoval(pld *payload.P2PNotaryRequest) {
+	if n.getAccount() == nil {
+		return
+	}
+
+	n.reqMtx.Lock()
+	defer n.reqMtx.Unlock()
+	r, ok := n.requests[pld.MainTransaction.Hash()]
+	if !ok {
+		return
+	}
+	for i, fb := range r.fallbacks {
+		if fb.Hash().Equals(pld.FallbackTransaction.Hash()) {
+			r.fallbacks = append(r.fallbacks[:i], r.fallbacks[i+1:]...)
+			break
+		}
+	}
+	if len(r.fallbacks) == 0 {
+		delete(n.requests, r.main.Hash())
+	}
+}
+
+// PostPersist is a callback which is called after new block is persisted.
+func (n *Notary) PostPersist(bc blockchainer.Blockchainer, pool *mempool.Pool, b *block.Block) {
+	if n.getAccount() == nil {
+		return
+	}
+
+	n.reqMtx.Lock()
+	defer n.reqMtx.Unlock()
+	for h, r := range n.requests {
+		if !r.isSent && r.typ != Unknown && r.nSigs == r.nSigsCollected && r.minNotValidBefore > bc.BlockHeight() {
+			if err := n.finalize(r.main); err != nil {
+				n.Config.Log.Error("failed to finalize main transaction", zap.Error(err))
+			} else {
+				r.isSent = true
+			}
+			continue
+		}
+		if r.minNotValidBefore <= bc.BlockHeight() { // then at least one of the fallbacks can already be sent.
+			newFallbacks := r.fallbacks[:0]
+			for _, fb := range r.fallbacks {
+				if nvb := fb.GetAttributes(transaction.NotValidBeforeT)[0].Value.(*transaction.NotValidBefore).Height; nvb <= bc.BlockHeight() {
+					if err := n.finalize(fb); err != nil {
+						newFallbacks = append(newFallbacks, fb) // wait for the next block to resend them
+					}
+				} else {
+					newFallbacks = append(newFallbacks, fb)
+				}
+			}
+			if len(newFallbacks) == 0 {
+				delete(n.requests, h)
+			} else {
+				r.fallbacks = newFallbacks
+			}
+		}
+	}
+}
+
+// finalize adds missing Notary witnesses to the transaction (main or fallback) and pushes it to the network.
+func (n *Notary) finalize(tx *transaction.Transaction) error {
+	acc := n.getAccount()
+	if acc == nil {
+		panic(errors.New("no available Notary account"))
+	}
+	notaryWitness := transaction.Witness{
+		InvocationScript:   append([]byte{byte(opcode.PUSHDATA1), 64}, acc.PrivateKey().Sign(tx.GetSignedPart())...),
+		VerificationScript: []byte{},
+	}
+	for i, signer := range tx.Signers {
+		if signer.Account == n.Config.Chain.GetNotaryContractScriptHash() {
+			tx.Scripts[i] = notaryWitness
+			break
+		}
+	}
+	return n.onTransaction(tx)
+}
+
+// verifyIncompleteWitnesses checks that tx either doesn't have all witnesses attached (in this case none of them
+// can be multisignature), or it only has a partial multisignature. It returns the request type (sig/multisig), the
+// number of signatures to be collected, sorted public keys (for multisig request only) and an error.
+func (n *Notary) verifyIncompleteWitnesses(tx *transaction.Transaction, nKeys uint8) (RequestType, uint8, keys.PublicKeys, error) {
+	var (
+		typ         RequestType
+		nSigs       int
+		nKeysActual uint8
+		pubsBytes   [][]byte
+		pubs        keys.PublicKeys
+		ok          bool
+	)
+	if len(tx.Signers) < 2 {
+		return Unknown, 0, nil, errors.New("transaction should have at least 2 signers")
+	}
+	if len(tx.Signers) != len(tx.Scripts) {
+		return Unknown, 0, nil, fmt.Errorf("transaction should have %d witnesses attached (completed + dummy)", len(tx.Signers))
+	}
+	if !tx.HasSigner(n.Config.Chain.GetNotaryContractScriptHash()) {
+		return Unknown, 0, nil, fmt.Errorf("P2PNotary contract should be a signer of the transaction")
+	}
+
+	for i, w := range tx.Scripts {
+		// do not check witness for Notary contract -- it will be replaced by proper witness in any case.
+		if tx.Signers[i].Account == n.Config.Chain.GetNotaryContractScriptHash() {
+			continue
+		}
+		if len(w.VerificationScript) == 0 {
+			// then it's a contract verification (can be combined with anything)
+			continue
+		}
+		if !tx.Signers[i].Account.Equals(hash.Hash160(w.VerificationScript)) { // https://github.com/nspcc-dev/neo-go/pull/1658#discussion_r564265987
+			return Unknown, 0, nil, fmt.Errorf("transaction should have valid verification script for signer #%d", i)
+		}
+		if nSigs, pubsBytes, ok = vm.ParseMultiSigContract(w.VerificationScript); ok {
+			if typ == Signature || typ == MultiSignature {
+				return Unknown, 0, nil, fmt.Errorf("bad type of witness #%d: only one multisignature witness is allowed", i)
+			}
+			typ = MultiSignature
+			nKeysActual = uint8(len(pubsBytes))
+			if len(w.InvocationScript) != 66 || !bytes.HasPrefix(w.InvocationScript, []byte{byte(opcode.PUSHDATA1), 64}) {
+				return Unknown, 0, nil, fmt.Errorf("multisignature invocation script should have length = 66 and be of the form [PUSHDATA1, 64, signatureBytes...]")
+			}
+			continue
+		}
+		if vm.IsSignatureContract(w.VerificationScript) {
+			if typ == MultiSignature {
+				return Unknown, 0, nil, fmt.Errorf("bad type of witness #%d: multisignature witness can not be combined with other witnesses", i)
+			}
+			typ = Signature
+			nSigs = int(nKeys)
+			continue
+		}
+		return Unknown, 0, nil, fmt.Errorf("unable to define the type of witness #%d", i)
+	}
+	switch typ {
+	case Signature:
+		if len(tx.Scripts) < int(nKeys+1) {
+			return Unknown, 0, nil, fmt.Errorf("transaction should comtain at least %d witnesses (1 for notary + nKeys)", nKeys+1)
+		}
+	case MultiSignature:
+		if nKeysActual != nKeys {
+			return Unknown, 0, nil, fmt.Errorf("bad m out of n partial multisignature witness: expected n = %d, got n = %d", nKeys, nKeysActual)
+		}
+		pubs = make(keys.PublicKeys, len(pubsBytes))
+		for i, pBytes := range pubsBytes {
+			pub, err := keys.NewPublicKeyFromBytes(pBytes, elliptic.P256())
+			if err != nil {
+				return Unknown, 0, nil, fmt.Errorf("invalid bytes of #%d public key: %s", i, hex.EncodeToString(pBytes))
+			}
+			pubs[i] = pub
+		}
+	default:
+		return Unknown, 0, nil, errors.New("unexpected Notary request type")
+	}
+	return typ, uint8(nSigs), pubs, nil
+}

pkg/services/notary/request_type.go

@@ -0,0 +1,13 @@
+package notary
+
+// RequestType represents the type of Notary request.
+type RequestType byte
+
+const (
+	// Unknown represents unknown request type which means that main tx witnesses are invalid.
+	Unknown RequestType = 0x00
+	// Signature represents standard single signature request type.
+	Signature RequestType = 0x01
+	// MultiSignature represents m out of n multisignature request type.
+	MultiSignature RequestType = 0x02
+)