frostfs-node/pkg/innerring/subnet.go

package innerring

import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"errors"
	"fmt"

	irsubnet "github.com/TrueCloudLab/frostfs-node/pkg/innerring/processors/subnet"
	netmapclient "github.com/TrueCloudLab/frostfs-node/pkg/morph/client/netmap"
	morphsubnet "github.com/TrueCloudLab/frostfs-node/pkg/morph/client/subnet"
	"github.com/TrueCloudLab/frostfs-node/pkg/morph/event"
	subnetevents "github.com/TrueCloudLab/frostfs-node/pkg/morph/event/subnet"
	"github.com/TrueCloudLab/frostfs-node/pkg/util"
	"github.com/TrueCloudLab/frostfs-sdk-go/netmap"
	"github.com/TrueCloudLab/frostfs-sdk-go/subnet"
	subnetid "github.com/TrueCloudLab/frostfs-sdk-go/subnet/id"
	"github.com/TrueCloudLab/frostfs-sdk-go/user"
	"github.com/nspcc-dev/neo-go/pkg/core/mempoolevent"
	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
	neogoutil "github.com/nspcc-dev/neo-go/pkg/util"
	"github.com/panjf2000/ants/v2"
	"go.uber.org/zap"
)

// IR server's component to handle Subnet contract notifications.
type subnetHandler struct {
	workerPool util.WorkerPool

	morphClient morphsubnet.Client

	putValidator irsubnet.PutValidator
}

// configuration of subnet component.
type subnetConfig struct {
	queueSize uint32
}

// makes IR server to catch Subnet notifications from the sidechain listener,
// and to release the corresponding processing queue on stop.
func (s *Server) initSubnet(cfg subnetConfig) {
	s.registerStarter(func() error {
		var err error

		// initialize queue for processing of the events from Subnet contract
		s.subnetHandler.workerPool, err = ants.NewPool(int(cfg.queueSize), ants.WithNonblocking(true))
		if err != nil {
			return fmt.Errorf("subnet queue initialization: %w", err)
		}

		// initialize morph client of Subnet contract
		clientMode := morphsubnet.NotaryAlphabet

		if s.sideNotaryConfig.disabled {
			clientMode = morphsubnet.NonNotary
		}

		var initPrm morphsubnet.InitPrm

		initPrm.SetBaseClient(s.morphClient)
		initPrm.SetContractAddress(s.contracts.subnet)
		initPrm.SetMode(clientMode)

		err = s.subnetHandler.morphClient.Init(initPrm)
		if err != nil {
			return fmt.Errorf("init morph subnet client: %w", err)
		}

		s.listenSubnet()

		return nil
	})

	s.registerCloser(func() error {
		s.stopSubnet()
		return nil
	})
}

// releases the Subnet contract notification processing queue.
func (s *Server) stopSubnet() {
	s.workerPool.Release()
}

// names of listened notification events from Subnet contract.
const (
	subnetCreateEvName       = "Put"
	subnetRemoveEvName       = "Delete"
	notarySubnetCreateEvName = "put"
)

// makes the IR server to listen to notifications of Subnet contract.
// All required resources must be initialized before (initSubnet).
// It works in one of two modes (configured): notary and non-notary.
//
// All handlers are executed only if the local node is an alphabet one.
//
// Events (notary):
//   - put (parser: subnetevents.ParseNotaryPut, handler: catchSubnetCreation);
//   - Delete (parser: subnetevents.ParseDelete, handler: catchSubnetCreation).
//
// Events (non-notary):
//   - Put (parser: subnetevents.ParsePut, handler: catchSubnetCreation);
//   - Delete (parser: subnetevents.ParseDelete, handler: catchSubnetCreation).
func (s *Server) listenSubnet() {
	if s.sideNotaryConfig.disabled {
		s.listenSubnetWithoutNotary()
		return
	}

	var (
		parserInfo  event.NotaryParserInfo
		handlerInfo event.NotaryHandlerInfo
	)

	parserInfo.SetScriptHash(s.contracts.subnet)
	handlerInfo.SetScriptHash(s.contracts.subnet)

	listenNotaryEvent := func(notifyName string, parser event.NotaryParser, handler event.Handler) {
		notifyTyp := event.NotaryTypeFromString(notifyName)

		parserInfo.SetMempoolType(mempoolevent.TransactionAdded)
		handlerInfo.SetMempoolType(mempoolevent.TransactionAdded)

		parserInfo.SetParser(parser)
		handlerInfo.SetHandler(handler)

		parserInfo.SetRequestType(notifyTyp)
		handlerInfo.SetRequestType(notifyTyp)

		s.morphListener.SetNotaryParser(parserInfo)
		s.morphListener.RegisterNotaryHandler(handlerInfo)
	}

	// subnet creation
	listenNotaryEvent(notarySubnetCreateEvName, subnetevents.ParseNotaryPut, s.onlyAlphabetEventHandler(s.catchSubnetCreation))
	// subnet removal
	listenNotifySubnetEvent(s, subnetRemoveEvName, subnetevents.ParseDelete, s.onlyAlphabetEventHandler(s.catchSubnetRemoval))
}

func (s *Server) listenSubnetWithoutNotary() {
	// subnet creation
	listenNotifySubnetEvent(s, subnetCreateEvName, subnetevents.ParsePut, s.onlyAlphabetEventHandler(s.catchSubnetCreation))
	// subnet removal
	listenNotifySubnetEvent(s, subnetRemoveEvName, subnetevents.ParseDelete, s.onlyAlphabetEventHandler(s.catchSubnetRemoval))
}

func listenNotifySubnetEvent(s *Server, notifyName string, parser event.NotificationParser, handler event.Handler) {
	var (
		parserInfo  event.NotificationParserInfo
		handlerInfo event.NotificationHandlerInfo
	)

	parserInfo.SetScriptHash(s.contracts.subnet)
	handlerInfo.SetScriptHash(s.contracts.subnet)

	notifyTyp := event.TypeFromString(notifyName)

	parserInfo.SetType(notifyTyp)
	handlerInfo.SetType(notifyTyp)

	parserInfo.SetParser(parser)
	handlerInfo.SetHandler(handler)

	s.morphListener.SetNotificationParser(parserInfo)
	s.morphListener.RegisterNotificationHandler(handlerInfo)
}

// catchSubnetCreation catches event of subnet creation from listener and queues the processing.
func (s *Server) catchSubnetCreation(e event.Event) {
	err := s.subnetHandler.workerPool.Submit(func() {
		s.handleSubnetCreation(e)
	})
	if err != nil {
		s.log.Error("subnet creation queue failure",
			zap.String("error", err.Error()),
		)
	}
}

// implements irsubnet.Put event interface required by irsubnet.PutValidator.
type putSubnetEvent struct {
	ev subnetevents.Put
}

// ReadID unmarshals the subnet ID from a binary FrostFS API protocol's format.
func (x putSubnetEvent) ReadID(id *subnetid.ID) error {
	return id.Unmarshal(x.ev.ID())
}

var errMissingSubnetOwner = errors.New("missing subnet owner")

// ReadCreator unmarshals the subnet creator from a binary FrostFS API protocol's format.
// Returns an error if the byte array is empty.
func (x putSubnetEvent) ReadCreator(id *user.ID) error {
	data := x.ev.Owner()

	if len(data) == 0 {
		return errMissingSubnetOwner
	}

	key, err := keys.NewPublicKeyFromBytes(data, elliptic.P256())
	if err != nil {
		return err
	}

	user.IDFromKey(id, (ecdsa.PublicKey)(*key))

	return nil
}

// ReadInfo unmarshal the subnet info from a binary FrostFS API protocol's format.
func (x putSubnetEvent) ReadInfo(info *subnet.Info) error {
	return info.Unmarshal(x.ev.Info())
}

// handleSubnetCreation handles an event of subnet creation parsed via subnetevents.ParsePut.
//
// Validates the event using irsubnet.PutValidator. Logs message about (dis)agreement.
func (s *Server) handleSubnetCreation(e event.Event) {
	putEv := e.(subnetevents.Put) // panic occurs only if we registered handler incorrectly

	err := s.subnetHandler.putValidator.Assert(putSubnetEvent{
		ev: putEv,
	})
	if err != nil {
		s.log.Info("discard subnet creation",
			zap.String("reason", err.Error()),
		)

		return
	}

	notaryMainTx := putEv.NotaryMainTx()

	isNotary := notaryMainTx != nil
	if isNotary {
		// re-sign notary request
		err = s.morphClient.NotarySignAndInvokeTX(notaryMainTx)
	} else {
		// send new transaction
		var prm morphsubnet.PutPrm

		prm.SetID(putEv.ID())
		prm.SetOwner(putEv.Owner())
		prm.SetInfo(putEv.Info())
		prm.SetTxHash(putEv.TxHash())

		_, err = s.subnetHandler.morphClient.Put(prm)
	}

	if err != nil {
		s.log.Error("approve subnet creation",
			zap.Bool("notary", isNotary),
			zap.String("error", err.Error()),
		)

		return
	}
}

// catchSubnetRemoval catches an event of subnet removal from listener and queues the processing.
func (s *Server) catchSubnetRemoval(e event.Event) {
	err := s.subnetHandler.workerPool.Submit(func() {
		s.handleSubnetRemoval(e)
	})
	if err != nil {
		s.log.Error("subnet removal handling failure",
			zap.String("error", err.Error()),
		)
	}
}

// handleSubnetRemoval handles event of subnet removal parsed via subnetevents.ParseDelete.
func (s *Server) handleSubnetRemoval(e event.Event) {
	delEv := e.(subnetevents.Delete) // panic occurs only if we registered handler incorrectly

	// handle subnet changes in netmap

	candidates, err := s.netmapClient.GetCandidates()
	if err != nil {
		s.log.Error("getting netmap candidates",
			zap.Error(err),
		)

		return
	}

	var removedID subnetid.ID
	err = removedID.Unmarshal(delEv.ID())
	if err != nil {
		s.log.Error("unmarshalling removed subnet ID",
			zap.String("error", err.Error()),
		)

		return
	}

	for i := range candidates {
		s.processCandidate(delEv.TxHash(), removedID, candidates[i])
	}
}

func (s *Server) processCandidate(txHash neogoutil.Uint256, removedID subnetid.ID, c netmap.NodeInfo) {
	removeSubnet := false
	log := s.log.With(
		zap.String("public_key", netmap.StringifyPublicKey(c)),
		zap.String("removed_subnet", removedID.String()),
	)

	err := c.IterateSubnets(func(id subnetid.ID) error {
		if removedID.Equals(id) {
			removeSubnet = true
			return netmap.ErrRemoveSubnet
		}

		return nil
	})
	if err != nil {
		log.Error("iterating node's subnets", zap.Error(err))
		log.Debug("removing node from netmap candidates")

		var updateStatePrm netmapclient.UpdatePeerPrm
		updateStatePrm.SetKey(c.PublicKey())
		updateStatePrm.SetHash(txHash)

		err = s.netmapClient.UpdatePeerState(updateStatePrm)
		if err != nil {
			log.Error("removing node from candidates",
				zap.Error(err),
			)
		}

		return
	}

	// remove subnet from node's information
	// if it contains removed subnet
	if removeSubnet {
		log.Debug("removing subnet from the node")

		var addPeerPrm netmapclient.AddPeerPrm
		addPeerPrm.SetNodeInfo(c)
		addPeerPrm.SetHash(txHash)

		err = s.netmapClient.AddPeer(addPeerPrm)
		if err != nil {
			log.Error("updating subnet info",
				zap.Error(err),
			)
		}
	}
}