certusone
/
wormhole
mirror of https://github.com/certusone/wormhole.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

node: (cleanup) remove injectC in favor of using msgC directly

This commit is contained in:
tbjump 2023-06-27 17:10:04 +00:00 committed by tbjump
parent 1d7976917a
commit 1d2e26c081
7 changed files with 39 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
node/pkg/adminrpc/adminserver.go
View File

@ -18,6 +18,8 @@ import (
"github.com/certusone/wormhole/node/pkg/watchers/evm/connectors"
"github.com/holiman/uint256"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"golang.org/x/exp/slices"
"github.com/certusone/wormhole/node/pkg/db"
@ -35,10 +37,18 @@ import (
sdktypes "github.com/cosmos/cosmos-sdk/types"
)
var (
vaaInjectionsTotal = promauto.NewCounter(
prometheus.CounterOpts{
Name: "wormhole_vaa_injections_total",
Help: "Total number of injected VAA queued for broadcast",
})
)
type nodePrivilegedService struct {
nodev1.UnimplementedNodePrivilegedServiceServer
db *db.Database
injectC chan<- *vaa.VAA
injectC chan<- *common.MessagePublication
obsvReqSendC chan<- *gossipv1.ObservationRequest
logger *zap.Logger
signedInC chan<- *gossipv1.SignedVAAWithQuorum
@ -52,7 +62,7 @@ type nodePrivilegedService struct {
func NewPrivService(
db *db.Database,
injectC chan<- *vaa.VAA,
injectC chan<- *common.MessagePublication,
obsvReqSendC chan<- *gossipv1.ObservationRequest,
logger *zap.Logger,
signedInC chan<- *gossipv1.SignedVAAWithQuorum,
@ -609,7 +619,19 @@ func (s *nodePrivilegedService) InjectGovernanceVAA(ctx context.Context, req *no
zap.String("digest", digest.String()),
)
s.injectC <- v
vaaInjectionsTotal.Inc()
s.injectC <- &common.MessagePublication{
TxHash: ethcommon.Hash{},
Timestamp: v.Timestamp,
Nonce: v.Nonce,
Sequence: v.Sequence,
ConsistencyLevel: v.ConsistencyLevel,
EmitterChain: v.EmitterChain,
EmitterAddress: v.EmitterAddress,
Payload: v.Payload,
Unreliable: false,
}
digests[i] = digest.Bytes()
}

3
node/pkg/node/adminServiceRunnable.go
View File

@ -18,7 +18,6 @@ import (
"github.com/certusone/wormhole/node/pkg/publicrpc"
"github.com/certusone/wormhole/node/pkg/supervisor"
"github.com/certusone/wormhole/node/pkg/watchers/evm/connectors"
"github.com/wormhole-foundation/wormhole/sdk/vaa"
"go.uber.org/zap"
ethcommon "github.com/ethereum/go-ethereum/common"
@ -28,7 +27,7 @@ import (
func adminServiceRunnable(
logger *zap.Logger,
socketPath string,
injectC chan<- *vaa.VAA,
injectC chan<- *common.MessagePublication,
signedInC chan<- *gossipv1.SignedVAAWithQuorum,
obsvReqSendC chan<- *gossipv1.ObservationRequest,
db *db.Database,

4
node/pkg/node/node.go
View File

@ -14,7 +14,6 @@ import (
"github.com/certusone/wormhole/node/pkg/reporter"
"github.com/certusone/wormhole/node/pkg/supervisor"
"github.com/wormhole-foundation/wormhole/sdk/vaa"
"go.uber.org/zap"
"google.golang.org/grpc"
)
@ -67,8 +66,6 @@ type G struct {
obsvReqC channelPair[*gossipv1.ObservationRequest]
// Outbound observation requests
obsvReqSendC channelPair[*gossipv1.ObservationRequest]
// Injected VAAs (manually generated rather than created via observation)
injectC channelPair[*vaa.VAA]
// acctC is the channel where messages will be put after they reached quorum in the accountant.
acctC channelPair[*common.MessagePublication]
}
@ -96,7 +93,6 @@ func (g *G) initializeBasic(logger *zap.Logger, rootCtxCancel context.CancelFunc
g.signedInC = makeChannelPair[*gossipv1.SignedVAAWithQuorum](inboundSignedVaaBufferSize)
g.obsvReqC = makeChannelPair[*gossipv1.ObservationRequest](observationRequestOutboundBufferSize)
g.obsvReqSendC = makeChannelPair[*gossipv1.ObservationRequest](observationRequestInboundBufferSize)
g.injectC = makeChannelPair[*vaa.VAA](0)
g.acctC = makeChannelPair[*common.MessagePublication](accountant.MsgChannelCapacity)
// Guardian set state managed by processor

11
node/pkg/node/options.go
View File

@ -286,6 +286,14 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
zap.Stringer("msgChainId", msg.EmitterChain),
zap.Stringer("watcherChainId", chainId),
)
} else if msg.EmitterAddress == vaa.GovernanceEmitter && msg.EmitterChain == vaa.GovernanceChain {
logger.Error(
"EMERGENCY: PLEASE REPORT THIS IMMEDIATELY! A Solana message was emitted from the governance emitter. This should never be possible.",
zap.Stringer("emitter_chain", msg.EmitterChain),
zap.Stringer("emitter_address", msg.EmitterAddress),
zap.Uint32("nonce", msg.Nonce),
zap.Stringer("txhash", msg.TxHash),
zap.Time("timestamp", msg.Timestamp))
} else {
g.msgC.writeC <- msg
}
@ -379,7 +387,7 @@ func GuardianOptionAdminService(socketPath string, ethRpc *string, ethContract *
adminService, err := adminServiceRunnable(
logger,
socketPath,
g.injectC.writeC,
g.msgC.writeC,
g.signedInC.writeC,
g.obsvReqSendC.writeC,
g.db,
@ -482,7 +490,6 @@ func GuardianOptionProcessor() *GuardianOption {
g.gossipSendC,
g.obsvC,
g.obsvReqSendC.writeC,
g.injectC.readC,
g.signedInC.readC,
g.gk,
g.gst,

46
node/pkg/processor/injection.go
View File

@ -1,46 +0,0 @@
package processor
import (
"context"
"encoding/hex"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/ethereum/go-ethereum/crypto"
"go.uber.org/zap"
"github.com/certusone/wormhole/node/pkg/supervisor"
"github.com/wormhole-foundation/wormhole/sdk/vaa"
)
var (
vaaInjectionsTotal = promauto.NewCounter(
prometheus.CounterOpts{
Name: "wormhole_vaa_injections_total",
Help: "Total number of injected VAA queued for broadcast",
})
)
// handleInjection processes a pre-populated VAA injected locally.
func (p *Processor) handleInjection(ctx context.Context, v *vaa.VAA) {
// Generate digest of the unsigned VAA.
digest := v.SigningDigest()
// The internal originator is responsible for logging the full VAA, just log the digest here.
supervisor.Logger(ctx).Info("signing injected VAA",
zap.String("digest", hex.EncodeToString(digest.Bytes())))
// Sign the digest using our node's guardian key.
s, err := crypto.Sign(digest.Bytes(), p.gk)
if err != nil {
panic(err)
}
p.logger.Info("observed and signed injected VAA",
zap.String("digest", hex.EncodeToString(digest.Bytes())),
zap.String("signature", hex.EncodeToString(s)))
vaaInjectionsTotal.Inc()
p.broadcastSignature(&VAA{VAA: *v}, s, nil)
}

16
node/pkg/processor/message.go
View File

@ -1,7 +1,6 @@
package processor
import (
"context"
"encoding/hex"
"github.com/mr-tron/base58"
@ -14,7 +13,6 @@ import (
"github.com/certusone/wormhole/node/pkg/common"
"github.com/certusone/wormhole/node/pkg/reporter"
"github.com/certusone/wormhole/node/pkg/supervisor"
"github.com/wormhole-foundation/wormhole/sdk/vaa"
)
@ -39,7 +37,7 @@ var (
// handleMessage processes a message received from a chain and instantiates our deterministic copy of the VAA. An
// event may be received multiple times and must be handled in an idempotent fashion.
func (p *Processor) handleMessage(ctx context.Context, k *common.MessagePublication) {
func (p *Processor) handleMessage(k *common.MessagePublication) {
if p.gs == nil {
p.logger.Warn("dropping observation since we haven't initialized our guardian set yet",
zap.Stringer("emitter_chain", k.EmitterChain),
@ -82,18 +80,6 @@ func (p *Processor) handleMessage(ctx context.Context, k *common.MessagePublicat
Unreliable: k.Unreliable,
}
// A governance message should never be emitted on-chain
if v.EmitterAddress == vaa.GovernanceEmitter && v.EmitterChain == vaa.GovernanceChain {
supervisor.Logger(ctx).Error(
"EMERGENCY: PLEASE REPORT THIS IMMEDIATELY! A Solana message was emitted from the governance emitter. This should never be possible.",
zap.Stringer("emitter_chain", k.EmitterChain),
zap.Stringer("emitter_address", k.EmitterAddress),
zap.Uint32("nonce", k.Nonce),
zap.Stringer("txhash", k.TxHash),
zap.Time("timestamp", k.Timestamp))
return
}
// Generate digest of the unsigned VAA.
digest := v.SigningDigest()

13
node/pkg/processor/processor.go
View File

@ -102,9 +102,6 @@ type Processor struct {
// signedInC is a channel of inbound signed VAA observations from p2p
signedInC <-chan *gossipv1.SignedVAAWithQuorum
// injectC is a channel of VAAs injected locally.
injectC <-chan *vaa.VAA
// gk is the node's guardian private key
gk *ecdsa.PrivateKey
@ -158,7 +155,6 @@ func NewProcessor(
gossipSendC chan<- []byte,
obsvC chan *common.MsgWithTimeStamp[gossipv1.SignedObservation],
obsvReqSendC chan<- *gossipv1.ObservationRequest,
injectC <-chan *vaa.VAA,
signedInC <-chan *gossipv1.SignedVAAWithQuorum,
gk *ecdsa.PrivateKey,
gst *common.GuardianSetState,
@ -176,7 +172,6 @@ func NewProcessor(
obsvC: obsvC,
obsvReqSendC: obsvReqSendC,
signedInC: signedInC,
injectC: injectC,
gk: gk,
gst: gst,
db: db,
@ -237,7 +232,7 @@ func (p *Processor) Run(ctx context.Context) error {
continue
}
}
p.handleMessage(ctx, k)
p.handleMessage(k)
case k := <-p.acctReadC:
if p.acct == nil {
@ -247,9 +242,7 @@ func (p *Processor) Run(ctx context.Context) error {
if !p.acct.IsMessageCoveredByAccountant(k) {
return fmt.Errorf("accountant published a message that is not covered by it: `%s`", k.MessageIDString())
}
p.handleMessage(ctx, k)
case v := <-p.injectC:
p.handleInjection(ctx, v)
p.handleMessage(k)
case m := <-p.obsvC:
observationChanDelay.Observe(float64(time.Since(m.Timestamp).Microseconds()))
p.handleObservation(ctx, m)
@ -280,7 +273,7 @@ func (p *Processor) Run(ctx context.Context) error {
continue
}
}
p.handleMessage(ctx, k)
p.handleMessage(k)
}
}
}