-
Notifications
You must be signed in to change notification settings - Fork 730
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
0e2ba62
commit fd94f23
Showing
13 changed files
with
714 additions
and
244 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| //nolint:unparam // this will be refactored in https://github.com/wormhole-foundation/wormhole/pull/1953 | ||
| package processor | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "github.com/certusone/wormhole/node/pkg/common" | ||
| "github.com/certusone/wormhole/node/pkg/p2p" | ||
| gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| // postObservationToBatch posts an individual observation to the batch processor. | ||
| func (p *Processor) postObservationToBatch(obs *gossipv1.Observation) { | ||
| select { | ||
| case p.batchObsvPubC <- obs: | ||
| default: | ||
| batchObservationChannelOverflow.WithLabelValues("batchObsvPub").Inc() | ||
| } | ||
| } | ||
|
|
||
| // batchProcessor is the entry point for the batch processor, which is responsible for taking individual | ||
| // observations and publishing them as batches. It limits the size of a batch and the delay before publishing. | ||
| func (p *Processor) batchProcessor(ctx context.Context) error { | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return nil | ||
| default: | ||
| if err := p.handleBatch(ctx); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // handleBatch reads observations from the channel, either until a timeout occurs or the batch is full. | ||
| // Then it builds a `SendObservationBatch` gossip message and posts it to p2p. | ||
| func (p *Processor) handleBatch(ctx context.Context) error { | ||
| ctx, cancel := context.WithTimeout(ctx, p2p.MaxObservationBatchDelay) | ||
| defer cancel() | ||
|
|
||
| observations, err := common.ReadFromChannelWithTimeout(ctx, p.batchObsvPubC, p2p.MaxObservationBatchSize) | ||
| if err != nil && !errors.Is(err, context.DeadlineExceeded) { | ||
| return fmt.Errorf("failed to read observations from the internal observation batch channel: %w", err) | ||
| } | ||
|
|
||
| if len(observations) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| batchMsg := gossipv1.SignedObservationBatch{ | ||
| Addr: p.ourAddr.Bytes(), | ||
| Observations: observations, | ||
| } | ||
|
|
||
| gossipMsg := gossipv1.GossipMessage{Message: &gossipv1.GossipMessage_SignedObservationBatch{SignedObservationBatch: &batchMsg}} | ||
| msg, err := proto.Marshal(&gossipMsg) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| select { | ||
| case p.gossipSendC <- msg: | ||
| default: | ||
| batchObservationChannelOverflow.WithLabelValues("gossipSend").Inc() | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package processor | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/certusone/wormhole/node/pkg/devnet" | ||
| "github.com/certusone/wormhole/node/pkg/p2p" | ||
| gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1" | ||
| ethcommon "github.com/ethereum/go-ethereum/common" | ||
| "github.com/ethereum/go-ethereum/crypto" | ||
| pubsub "github.com/libp2p/go-libp2p-pubsub" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/wormhole-foundation/wormhole/sdk/vaa" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| func getUniqueVAA(seqNo uint64) vaa.VAA { | ||
| var payload = []byte{97, 97, 97, 97, 97, 97} | ||
| var governanceEmitter = vaa.Address{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 4} | ||
|
|
||
| vaa := vaa.VAA{ | ||
| Version: uint8(1), | ||
| GuardianSetIndex: uint32(1), | ||
| Signatures: nil, | ||
| Timestamp: time.Unix(0, 0), | ||
| Nonce: uint32(1), | ||
| Sequence: seqNo, | ||
| ConsistencyLevel: uint8(32), | ||
| EmitterChain: vaa.ChainIDSolana, | ||
| EmitterAddress: governanceEmitter, | ||
| Payload: payload, | ||
| } | ||
|
|
||
| return vaa | ||
| } | ||
|
|
||
| func TestMarshalSignedObservationBatch(t *testing.T) { | ||
| gk := devnet.InsecureDeterministicEcdsaKeyByIndex(crypto.S256(), uint64(0)) | ||
| require.NotNil(t, gk) | ||
|
|
||
| NumObservations := uint64(p2p.MaxObservationBatchSize) | ||
| observations := make([]*gossipv1.Observation, 0, NumObservations) | ||
| for seqNo := uint64(1); seqNo <= NumObservations; seqNo++ { | ||
| vaa := getUniqueVAA(seqNo) | ||
| digest := vaa.SigningDigest() | ||
| sig, err := crypto.Sign(digest.Bytes(), gk) | ||
| require.NoError(t, err) | ||
|
|
||
| observations = append(observations, &gossipv1.Observation{ | ||
| Hash: digest.Bytes(), | ||
| Signature: sig, | ||
| TxHash: ethcommon.HexToHash("0x06f541f5ecfc43407c31587aa6ac3a689e8960f36dc23c332db5510dfc6a4063").Bytes(), | ||
| MessageId: vaa.MessageID(), | ||
| }) | ||
| } | ||
|
|
||
| obsBuf, err := proto.Marshal(observations[0]) | ||
| require.NoError(t, err) | ||
| assert.Equal(t, 205, len(obsBuf)) | ||
|
|
||
| batch := gossipv1.SignedObservationBatch{ | ||
| Addr: crypto.PubkeyToAddress(gk.PublicKey).Bytes(), | ||
| Observations: observations, | ||
| } | ||
|
|
||
| buf, err := proto.Marshal((&batch)) | ||
| require.NoError(t, err) | ||
| assert.Greater(t, pubsub.DefaultMaxMessageSize, len(buf)) | ||
|
|
||
| var batch2 gossipv1.SignedObservationBatch | ||
| err = proto.Unmarshal(buf, &batch2) | ||
| require.NoError(t, err) | ||
|
|
||
| assert.True(t, bytes.Equal(batch.Addr, batch2.Addr)) | ||
| assert.Equal(t, len(batch.Observations), len(batch2.Observations)) | ||
| for idx := range batch2.Observations { | ||
| assert.True(t, bytes.Equal(batch.Observations[idx].Hash, batch2.Observations[idx].Hash)) | ||
| assert.True(t, bytes.Equal(batch.Observations[idx].Signature, batch2.Observations[idx].Signature)) | ||
| assert.True(t, bytes.Equal(batch.Observations[idx].TxHash, batch2.Observations[idx].TxHash)) | ||
| assert.Equal(t, batch.Observations[idx].MessageId, batch2.Observations[idx].MessageId) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.