-
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
c2496cd
commit 601ffc8
Showing
14 changed files
with
745 additions
and
268 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
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 @@ | ||
| 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) { | ||
| // TODO: This will be enabled as part of the gossip split PR. | ||
| // 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) | ||
| } | ||
| } |
Oops, something went wrong.