realloc/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Releases Packages Activity

[#971] *: Add notification TX hash to neofs/netmap morph client calls

Browse source 
Add hash of the TX that generated notification
to neofs/netmap event structures. Adapt all
neofs/netmap wrapper calls to new structures.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2021-11-10 14:05:51 +03:00 • committed by Alex Vanin
parent c25f5a86ae
commit bad739258e
19 changed files with 220 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

15
pkg/innerring/processors/balance/process_assets.go
View file

@ -1,6 +1,7 @@
package balance
import (
neofscontract "github.com/nspcc-dev/neofs-node/pkg/morph/client/neofs/wrapper"
balanceEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/balance"
"go.uber.org/zap"
)
@ -13,11 +14,15 @@ func (bp *Processor) processLock(lock *balanceEvent.Lock) {
return
}
err := bp.neofsClient.Cheque(
lock.ID(),
lock.User(),
bp.converter.ToFixed8(lock.Amount()),
lock.LockAccount())
prm := neofscontract.ChequePrm{}
prm.SetID(lock.ID())
prm.SetUser(lock.User())
prm.SetAmount(bp.converter.ToFixed8(lock.Amount()))
prm.SetLock(lock.LockAccount())
prm.SetHash(lock.TxHash())
err := bp.neofsClient.Cheque(prm)
if err != nil {
bp.log.Error("can't send lock asset tx", zap.Error(err))
}

23
pkg/innerring/processors/governance/events.go
View file

@ -1,11 +1,26 @@
package governance
// Sync is a event to start governance synchronization.
type Sync struct{}
import "github.com/nspcc-dev/neo-go/pkg/util"
// Sync is an event to start governance synchronization.
type Sync struct {
// txHash is used in notary environmental
// for calculating unique but same for
// all notification receivers values.
txHash util.Uint256
}
// TxHash returns hash of the TX that triggers
// synchronization process.
func (s Sync) TxHash() util.Uint256 {
return s.txHash
}
// MorphEvent implements Event interface.
func (s Sync) MorphEvent() {}
func NewSyncEvent() Sync {
return Sync{}
// NewSyncEvent creates Sync event that was produced
// in transaction with txHash hash.
func NewSyncEvent(txHash util.Uint256) Sync {
return Sync{txHash: txHash}
}

11
pkg/innerring/processors/governance/handlers.go
View file

@ -3,22 +3,29 @@ package governance
import (
"github.com/nspcc-dev/neo-go/pkg/core/native"
"github.com/nspcc-dev/neo-go/pkg/core/native/noderoles"
"github.com/nspcc-dev/neo-go/pkg/util"
"github.com/nspcc-dev/neofs-node/pkg/morph/event"
"github.com/nspcc-dev/neofs-node/pkg/morph/event/rolemanagement"
"go.uber.org/zap"
)
func (gp *Processor) HandleAlphabetSync(e event.Event) {
var typ string
var (
typ string
hash util.Uint256
)
switch et := e.(type) {
case Sync:
typ = "sync"
hash = et.TxHash()
case rolemanagement.Designate:
if et.Role != noderoles.NeoFSAlphabet {
return
}
typ = native.DesignationEventName
hash = et.TxHash
default:
return
}
@ -27,7 +34,7 @@ func (gp *Processor) HandleAlphabetSync(e event.Event) {
// send event to the worker pool
err := gp.pool.Submit(func() { gp.processAlphabetSync() })
err := gp.pool.Submit(func() { gp.processAlphabetSync(hash) })
if err != nil {
// there system can be moved into controlled degradation stage
gp.log.Warn("governance worker pool drained",

43
pkg/innerring/processors/governance/process_update.go
View file

@ -6,7 +6,12 @@ import (
"sort"
"strings"
"github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/nspcc-dev/neo-go/pkg/util"
"github.com/nspcc-dev/neofs-node/pkg/morph/client"
neofscontract "github.com/nspcc-dev/neofs-node/pkg/morph/client/neofs/wrapper"
"go.uber.org/zap"
)
@ -14,7 +19,7 @@ const (
alphabetUpdateIDPrefix = "AlphabetUpdate"
)
func (gp *Processor) processAlphabetSync() {
func (gp *Processor) processAlphabetSync(txHash util.Uint256) {
if !gp.alphabetState.IsAlphabet() {
gp.log.Info("non alphabet mode, ignore alphabet sync")
return
@ -51,8 +56,13 @@ func (gp *Processor) processAlphabetSync() {
zap.String("new_alphabet", prettyKeys(newAlphabet)),
)
votePrm := VoteValidatorPrm{
Validators: newAlphabet,
Hash: &txHash,
}
// 1. Vote to side chain committee via alphabet contracts.
err = gp.voter.VoteForSidechainValidator(newAlphabet)
err = gp.voter.VoteForSidechainValidator(votePrm)
if err != nil {
gp.log.Error("can't vote for side chain committee",
zap.String("error", err.Error()))
@ -77,9 +87,19 @@ func (gp *Processor) processAlphabetSync() {
)
if gp.notaryDisabled {
err = gp.netmapClient.UpdateInnerRing(newInnerRing)
updPrm := wrapper.UpdateIRPrm{}
updPrm.SetKeys(newInnerRing)
updPrm.SetHash(txHash)
err = gp.netmapClient.UpdateInnerRing(updPrm)
} else {
err = gp.morphClient.UpdateNeoFSAlphabetList(newInnerRing)
updPrm := client.UpdateAlphabetListPrm{}
updPrm.SetList(newInnerRing)
updPrm.SetHash(txHash)
err = gp.morphClient.UpdateNeoFSAlphabetList(updPrm)
}
if err != nil {
@ -91,7 +111,13 @@ func (gp *Processor) processAlphabetSync() {
if !gp.notaryDisabled {
// 3. Update notary role in side chain.
err = gp.morphClient.UpdateNotaryList(newAlphabet)
updPrm := client.UpdateNotaryListPrm{}
updPrm.SetList(newAlphabet)
updPrm.SetHash(txHash)
err = gp.morphClient.UpdateNotaryList(updPrm)
if err != nil {
gp.log.Error("can't update list of notary nodes in side chain",
zap.String("error", err.Error()))
@ -106,7 +132,12 @@ func (gp *Processor) processAlphabetSync() {
id := append([]byte(alphabetUpdateIDPrefix), buf...)
err = gp.neofsClient.AlphabetUpdate(id, newAlphabet)
prm := neofscontract.AlphabetUpdatePrm{}
prm.SetID(id)
prm.SetPubs(newAlphabet)
err = gp.neofsClient.AlphabetUpdate(prm)
if err != nil {
gp.log.Error("can't update list of alphabet nodes in neofs contract",
zap.String("error", err.Error()))

17
pkg/innerring/processors/governance/processor.go
View file

@ -26,12 +26,21 @@ type (
AlphabetState interface {
IsAlphabet() bool
}
)
// Voter is a callback interface for alphabet contract voting.
Voter interface {
VoteForSidechainValidator(keys keys.PublicKeys) error
}
// VoteValidatorPrm groups parameters of the VoteForSidechainValidator
// operation.
type VoteValidatorPrm struct {
Validators keys.PublicKeys
Hash *util.Uint256 // hash of the transaction that triggered voting
}
// Voter is a callback interface for alphabet contract voting.
type Voter interface {
VoteForSidechainValidator(VoteValidatorPrm) error
}
type (
// EpochState is a callback interface for innerring global state.
EpochState interface {
EpochCounter() uint64

10
pkg/innerring/processors/neofs/process_config.go
View file

@ -1,6 +1,7 @@
package neofs
import (
"github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper"
neofsEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/neofs"
"go.uber.org/zap"
)
@ -13,7 +14,14 @@ func (np *Processor) processConfig(config *neofsEvent.Config) {
return
}
err := np.netmapClient.SetConfig(config.ID(), config.Key(), config.Value())
prm := wrapper.SetConfigPrm{}
prm.SetID(config.ID())
prm.SetKey(config.Key())
prm.SetValue(config.Value())
prm.SetHash(config.TxHash())
err := np.netmapClient.SetConfig(prm)
if err != nil {
np.log.Error("can't relay set config event", zap.Error(err))
}

13
pkg/innerring/processors/netmap/internal_events.go
View file

@ -1,8 +1,21 @@
package netmap
import "github.com/nspcc-dev/neo-go/pkg/util"
// netmapCleanupTick is a event to remove offline nodes.
type netmapCleanupTick struct {
epoch uint64
// txHash is used in notary environmental
// for calculating unique but same for
// all notification receivers values.
txHash util.Uint256
}
// TxHash returns hash of the TX that triggers
// synchronization process.
func (s netmapCleanupTick) TxHash() util.Uint256 {
return s.txHash
}
// MorphEvent implements Event interface.

21
pkg/innerring/processors/netmap/process_cleanup.go
View file

@ -2,7 +2,7 @@ package netmap
import (
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
netmapEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/netmap"
netmapclient "github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper"
"github.com/nspcc-dev/neofs-sdk-go/netmap"
"go.uber.org/zap"
)
@ -25,20 +25,13 @@ func (np *Processor) processNetmapCleanupTick(ev netmapCleanupTick) {
np.log.Info("vote to remove node from netmap", zap.String("key", s))
if np.notaryDisabled {
err = np.netmapClient.UpdatePeerState(key.Bytes(), netmap.NodeStateOffline)
} else {
// use epoch as TX nonce to prevent collisions
err = np.netmapClient.Morph().NotaryInvoke(
np.netmapClient.ContractAddress(),
0,
uint32(ev.epoch),
netmapEvent.UpdateStateNotaryEvent,
int64(netmap.NodeStateOffline.ToV2()),
key.Bytes(),
)
}
prm := netmapclient.UpdatePeerPrm{}
prm.SetKey(key.Bytes())
prm.SetState(netmap.NodeStateOffline)
prm.SetHash(ev.TxHash())
err = np.netmapClient.UpdatePeerState(prm)
if err != nil {
np.log.Error("can't invoke netmap.UpdateState", zap.Error(err))
}

10
pkg/innerring/processors/netmap/process_epoch.go
View file

@ -10,8 +10,8 @@ import (
// Process new epoch notification by setting global epoch value and resetting
// local epoch timer.
func (np *Processor) processNewEpoch(event netmapEvent.NewEpoch) {
epoch := event.EpochNumber()
func (np *Processor) processNewEpoch(ev netmapEvent.NewEpoch) {
epoch := ev.EpochNumber()
epochDuration, err := np.netmapClient.EpochDuration()
if err != nil {
@ -47,11 +47,11 @@ func (np *Processor) processNewEpoch(event netmapEvent.NewEpoch) {
}
np.netmapSnapshot.update(networkMap, epoch)
np.handleCleanupTick(netmapCleanupTick{epoch: epoch})
np.handleCleanupTick(netmapCleanupTick{epoch: epoch, txHash: ev.TxHash()})
np.handleNewAudit(audit.NewAuditStartEvent(epoch))
np.handleAuditSettlements(settlement.NewAuditEvent(epoch))
np.handleAlphabetSync(governance.NewSyncEvent())
np.handleNotaryDeposit(event)
np.handleAlphabetSync(governance.NewSyncEvent(ev.TxHash()))
np.handleNotaryDeposit(ev)
}
// Process new epoch tick by invoking new epoch method in network map contract.

14
pkg/innerring/processors/netmap/process_peers.go
View file

@ -5,6 +5,7 @@ import (
"sort"
"strings"
netmapclient "github.com/nspcc-dev/neofs-node/pkg/morph/client/netmap/wrapper"
netmapEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/netmap"
"github.com/nspcc-dev/neofs-sdk-go/netmap"
"go.uber.org/zap"
@ -81,18 +82,22 @@ func (np *Processor) processAddPeer(ev netmapEvent.AddPeer) {
np.log.Info("approving network map candidate",
zap.String("key", keyString))
prm := netmapclient.AddPeerPrm{}
prm.SetNodeInfo(nodeInfo)
if nr := ev.NotaryRequest(); nr != nil {
// create new notary request with the original nonce
err = np.netmapClient.Morph().NotaryInvoke(
np.netmapClient.ContractAddress(),
0,
nr.MainTransaction.Nonce,
nil,
netmapEvent.AddPeerNotaryEvent,
nodeInfoBinary,
)
} else {
// notification event case
err = np.netmapClient.AddPeer(nodeInfo)
err = np.netmapClient.AddPeer(prm)
}
if err != nil {
@ -126,7 +131,12 @@ func (np *Processor) processUpdatePeer(ev netmapEvent.UpdatePeer) {
if nr := ev.NotaryRequest(); nr != nil {
err = np.netmapClient.Morph().NotarySignAndInvokeTX(nr.MainTransaction)
} else {
err = np.netmapClient.UpdatePeerState(ev.PublicKey().Bytes(), ev.Status())
prm := netmapclient.UpdatePeerPrm{}
prm.SetState(ev.Status())
prm.SetKey(ev.PublicKey().Bytes())
err = np.netmapClient.UpdatePeerState(prm)
}
if err != nil {
np.log.Error("can't invoke netmap.UpdatePeer", zap.Error(err))