ceremonyclient/node/consensus/master/peer_messaging.go

package master

import (
	"bytes"

	"github.com/pkg/errors"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
	"source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb"
	"source.quilibrium.com/quilibrium/monorepo/node/protobufs"
)

func (e *MasterClockConsensusEngine) handleSync(message *pb.Message) error {
	e.logger.Debug(
		"received peer message",
		zap.Binary("data", message.Data),
		zap.Binary("from", message.From),
		zap.Binary("signature", message.Signature),
	)
	msg := &protobufs.Message{}

	if err := proto.Unmarshal(message.Data, msg); err != nil {
		return errors.Wrap(err, "handle sync")
	}

	any := &anypb.Any{}
	if err := proto.Unmarshal(msg.Payload, any); err != nil {
		return errors.Wrap(err, "handle sync")
	}

	eg := errgroup.Group{}
	eg.SetLimit(len(e.executionEngines))

	for name := range e.executionEngines {
		name := name
		eg.Go(func() error {
			applications := e.executionEngines[name].GetSupportedApplications()
			for _, application := range applications {
				if bytes.Equal(msg.Address, application.Address) {
					messages, err := e.executionEngines[name].ProcessMessage(
						msg.Address,
						msg,
					)
					if err != nil {
						e.logger.Error(
							"could not process message for engine",
							zap.Error(err),
							zap.String("engine_name", name),
						)
						return errors.Wrap(err, "handle message")
					}

					for _, m := range messages {
						if err := e.publishMessage(e.filter, m); err != nil {
							e.logger.Error(
								"could not publish message for engine",
								zap.Error(err),
								zap.String("engine_name", name),
							)
							return errors.Wrap(err, "handle message")
						}
					}
				}
			}

			return nil
		})
	}

	if err := eg.Wait(); err != nil {
		e.logger.Error("rejecting invalid message", zap.Error(err))
		return errors.Wrap(err, "handle sync")
	}

	switch any.TypeUrl {
	case protobufs.ClockFramesResponseType:
		if err := e.handleClockFramesResponse(
			message.From,
			any,
		); err != nil {
			return errors.Wrap(err, "handle sync")
		}
	case protobufs.ClockFramesRequestType:
		if err := e.handleClockFramesRequest(
			message.From,
			any,
		); err != nil {
			return errors.Wrap(err, "handle sync")
		}
	}

	return nil
}

func (e *MasterClockConsensusEngine) createPeerReceiveChannel(
	peerID []byte,
) []byte {
	return append(append([]byte{}, peerID...), e.pubSub.GetPeerID()...)
}

func (e *MasterClockConsensusEngine) createPeerSendChannel(
	peerID []byte,
) []byte {
	return append(append([]byte{}, e.pubSub.GetPeerID()...), peerID...)
}

func (e *MasterClockConsensusEngine) handleClockFramesResponse(
	peerID []byte,
	any *anypb.Any,
) error {
	if bytes.Equal(peerID, e.pubSub.GetPeerID()) {
		return nil
	}

	if !bytes.Equal(peerID, e.syncingTarget) {
		e.logger.Warn(
			"received clock frames response from unexpected target",
			zap.Binary("peer_id", peerID),
			zap.Binary("expected_peer_id", e.syncingTarget),
		)
		return nil
	}

	e.syncingStatus = SyncStatusSynchronizing

	defer func() { e.syncingStatus = SyncStatusNotSyncing }()

	response := &protobufs.ClockFramesResponse{}
	if err := any.UnmarshalTo(response); err != nil {
		return errors.Wrap(err, "handle clock frames response")
	}

	for _, frame := range response.ClockFrames {
		e.logger.Info(
			"processing clock frame",
			zap.Binary("sender", peerID),
			zap.Binary("filter", frame.Filter),
			zap.Uint64("frame_number", frame.FrameNumber),
		)

		if err := frame.VerifyMasterClockFrame(); err != nil {
			e.logger.Error("could not verify clock frame", zap.Error(err))
			return errors.Wrap(err, "handle clock frame response")
		}

		e.logger.Info(
			"clock frame was valid",
			zap.Binary("sender", peerID),
			zap.Binary("filter", frame.Filter),
			zap.Uint64("frame_number", frame.FrameNumber),
		)

		if e.frame < frame.FrameNumber {
			if err := e.enqueueSeenFrame(frame); err != nil {
				e.logger.Error("could not enqueue seen clock frame", zap.Error(err))
				return errors.Wrap(err, "handle clock frame response")
			}
		}
	}

	return nil
}

func (e *MasterClockConsensusEngine) handleClockFramesRequest(
	peerID []byte,
	any *anypb.Any,
) error {
	if bytes.Equal(peerID, e.pubSub.GetPeerID()) {
		return nil
	}

	request := &protobufs.ClockFramesRequest{}
	if err := any.UnmarshalTo(request); err != nil {
		return errors.Wrap(err, "handle clock frame request")
	}

	channel := e.createPeerSendChannel(peerID)

	e.pubSub.Subscribe(channel, e.handleSync, true)

	e.logger.Info(
		"received clock frame request",
		zap.Binary("peer_id", peerID),
		zap.Uint64("from_frame_number", request.FromFrameNumber),
		zap.Uint64("to_frame_number", request.ToFrameNumber),
	)

	from := request.FromFrameNumber

	if e.frame < from || len(e.historicFrames) == 0 {
		e.logger.Info(
			"peer asked for undiscovered frame",
			zap.Binary("peer_id", peerID),
			zap.Uint64("frame_number", request.FromFrameNumber),
		)

		if err := e.publishMessage(channel, &protobufs.ClockFramesResponse{
			Filter:          request.Filter,
			FromFrameNumber: 0,
			ToFrameNumber:   0,
			ClockFrames:     []*protobufs.ClockFrame{},
		}); err != nil {
			return errors.Wrap(err, "handle clock frame request")
		}

		return nil
	}

	to := request.ToFrameNumber
	if to == 0 || to-request.FromFrameNumber > 128 {
		to = request.FromFrameNumber + 127
	}

	if int(to) > len(e.historicFrames)-1 {
		to = uint64(len(e.historicFrames) - 1)
	}

	e.logger.Info(
		"sending response",
		zap.Binary("peer_id", peerID),
		zap.Uint64("from", from),
		zap.Uint64("to", to),
		zap.Uint64("total_frames", uint64(to-from+1)),
	)

	if err := e.publishMessage(channel, &protobufs.ClockFramesResponse{
		Filter:          request.Filter,
		FromFrameNumber: request.FromFrameNumber,
		ToFrameNumber:   to,
		ClockFrames:     e.historicFrames[from:to],
	}); err != nil {
		return errors.Wrap(err, "handle clock frame request")
	}

	return nil
}