node: broadcast a SignedVAAWithQuorum message when a VAA reaches quorum

Change-Id: I5dbefcbcf146bda6e3ef9607a7b49ca1cdf5321d

node/pkg/processor/broadcast.go

@@ -63,3 +63,21 @@ func (p *Processor) broadcastSignature(v *vaa.VAA, signature []byte) {
 
 	observationsBroadcastTotal.Inc()
 }
+
+func (p *Processor) broadcastSignedVAA(v *vaa.VAA) {
+	b, err := v.Marshal()
+	if err != nil {
+		panic(err)
+	}
+
+	w := gossipv1.GossipMessage{Message: &gossipv1.GossipMessage_SignedVaaWithQuorum{
+		SignedVaaWithQuorum: &gossipv1.SignedVAAWithQuorum{Vaa: b},
+	}}
+
+	msg, err := proto.Marshal(&w)
+	if err != nil {
+		panic(err)
+	}
+
+	p.sendC <- msg
+}

node/pkg/processor/observation.go

@@ -240,6 +240,8 @@ func (p *Processor) handleObservation(ctx context.Context, m *gossipv1.SignedObs
 			if err := p.db.StoreSignedVAA(signed); err != nil {
 				p.logger.Error("failed to store signed VAA", zap.Error(err))
 			}
+
+			p.broadcastSignedVAA(signed)
 			p.attestationEvents.ReportVAAQuorum(signed)
 			p.state.vaaSignatures[hash].submitted = true
 		} else {

proto/gossip/v1/gossip.proto

@@ -10,6 +10,7 @@ message GossipMessage {
     Heartbeat heartbeat = 1;
     SignedObservation signed_observation = 2;
     SignedHeartbeat signed_heartbeat = 3;
+    SignedVAAWithQuorum signed_vaa_with_quorum = 4;
   }
 }
 
@@ -78,3 +79,10 @@ message SignedObservation {
   // ECSDA signature of the hash using the node's guardian key.
   bytes signature = 3;
 }
+
+// A SignedVAAWithQuorum message is sent by nodes whenever one of the VAAs they observed
+// reached a 2/3+ quorum to be considered valid. Signed VAAs are broadcasted to the gossip
+// network to allow nodes to persist them even if they failed to observe the signature.
+message SignedVAAWithQuorum {
+  bytes vaa = 1;
+}