discovery: send signature announcement once

This change aligns our behavior with latest protocol guidelines. That
is, when we receive a signature announcement when we already have the
full proof we reply with our signature announcement once per
(re)connection.

see: lightning/bolts#1256

Author: gijswijs

discovery/gossiper.go

@@ -473,6 +473,60 @@ func (c *cachedReject) Size() (uint64, error) {
 	return 1, nil
 }
 
+// sentAnnSigsTracker tracks which announcement signatures we've sent to which
+// peers.
+type sentAnnSigsTracker struct {
+	mu   sync.Mutex
+	sigs map[route.Vertex]map[lnwire.ChannelID]struct{}
+}
+
+// newSentAnnSigsTracker creates a new sentAnnSigsTracker.
+func newSentAnnSigsTracker() *sentAnnSigsTracker {
+	return &sentAnnSigsTracker{
+		sigs: make(map[route.Vertex]map[lnwire.ChannelID]struct{}),
+	}
+}
+
+// deletePeer deletes all records of sent announcement signatures for a given
+// peer.
+func (s *sentAnnSigsTracker) deletePeer(peer route.Vertex) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.sigs, peer)
+}
+
+// sentToPeer returns true if we have already sent the announcement signature
+// for a given channel to a peer.
+func (s *sentAnnSigsTracker) sentToPeer(peer route.Vertex,
+	cid lnwire.ChannelID) bool {
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	sentChans, ok := s.sigs[peer]
+	if !ok {
+		return false
+	}
+
+	_, ok = sentChans[cid]
+
+	return ok
+}
+
+// addToPeer adds a record of sending an announcement signature for a given
+// channel to a peer.
+func (s *sentAnnSigsTracker) addToPeer(peer route.Vertex,
+	cid lnwire.ChannelID) {
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.sigs[peer]; !ok {
+		s.sigs[peer] = make(map[lnwire.ChannelID]struct{})
+	}
+	s.sigs[peer][cid] = struct{}{}
+}
+
 // AuthenticatedGossiper is a subsystem which is responsible for receiving
 // announcements, validating them and applying the changes to router, syncing
 // lightning network with newly connected nodes, broadcasting announcements
@@ -568,6 +622,11 @@ type AuthenticatedGossiper struct {
 	// vb is used to enforce job dependency ordering of gossip messages.
 	vb *ValidationBarrier
 
+	// sentAnnSigs tracks which announcement signatures we've sent to which
+	// peers. We'll use this to ensure we don't re-send the same signatures
+	// to a peer during a single connection.
+	sentAnnSigs *sentAnnSigsTracker
+
 	sync.Mutex
 
 	cancel fn.Option[context.CancelFunc]
@@ -595,6 +654,7 @@ func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper
 		),
 		chanUpdateRateLimiter: make(map[uint64][2]*rate.Limiter),
 		banman:                newBanman(cfg.BanThreshold),
+		sentAnnSigs:           newSentAnnSigsTracker(),
 	}
 
 	gossiper.vb = NewValidationBarrier(1000, gossiper.quit)
@@ -1677,6 +1737,14 @@ func (d *AuthenticatedGossiper) handleNetworkMessages(ctx context.Context,
 // queries. We'll allocate a new gossip syncer for it, and start any goroutines
 // needed to handle new queries.
 func (d *AuthenticatedGossiper) InitSyncState(syncPeer lnpeer.Peer) {
+	peerVertex := route.NewVertex(syncPeer.IdentityKey())
+
+	// We keep track if we already have sent an announcement signature to
+	// each peer during a connection, so we don't send it twice. If a peer
+	// dis- and reconnects we clear the record. If the peer connects for the
+	// first time this is a no-op.
+	d.sentAnnSigs.deletePeer(peerVertex)
+
 	d.syncMgr.InitSyncState(syncPeer)
 }
 
@@ -3542,12 +3610,26 @@ func (d *AuthenticatedGossiper) handleAnnSig(ctx context.Context,
 	if chanInfo.AuthProof != nil {
 		// If we already have the fully assembled proof, then the peer
 		// sending us their proof has probably not received our local
-		// proof yet. So be kind and send them our proof.
+		// proof yet. So be kind and send them our proof, but only if we
+		// haven't done so since (re)connecting.
 		if nMsg.isRemote {
-			peerID := nMsg.source.SerializeCompressed()
-			log.Debugf("Got AnnounceSignatures for channel with " +
-				"full proof.")
+			peerVertex := route.NewVertex(nMsg.source)
+
+			if d.sentAnnSigs.sentToPeer(peerVertex, ann.ChannelID) {
+				log.Debugf("Skipping sending " +
+					"announcement signatures " +
+					"since we already did during " +
+					"this connection.")
 
+				nMsg.err <- nil
+
+				return nil, true
+			}
+
+			peerID := nMsg.source.SerializeCompressed()
+			peerIsFirstNode := bytes.Equal(
+				peerID, chanInfo.NodeKey1Bytes[:],
+			)
 			d.wg.Add(1)
 			go func() {
 				defer d.wg.Done()
@@ -3558,28 +3640,39 @@ func (d *AuthenticatedGossiper) handleAnnSig(ctx context.Context,
 					ann.ChannelID, peerID)
 
 				chanAP := chanInfo.AuthProof
-				var remoteNSB []byte
-				var remoteBSB []byte
-				if isFirstNode {
-					remoteNSB = chanAP.NodeSig2Bytes
-					remoteBSB = chanAP.BitcoinSig2Bytes
+				// Pick the local signatures that the peer does
+				// NOT have yet. If the peer is node1, they sent
+				// us node1/bitcoin1 already, so we respond with
+				// node2/bitcoin2. Otherwise we respond with
+				// node1/bitcoin1.
+				var localNSB, localBSB []byte
+				if peerIsFirstNode {
+					localNSB = chanAP.NodeSig2Bytes
+					localBSB = chanAP.BitcoinSig2Bytes
 				} else {
-					remoteNSB = chanAP.NodeSig1Bytes
-					remoteBSB = chanAP.BitcoinSig1Bytes
+					localNSB = chanAP.NodeSig1Bytes
+					localBSB = chanAP.BitcoinSig1Bytes
 				}
 
+				// Construct an AnnounceSignatures message from
+				// the raw signature material. This gives the
+				// peer our half of the proof so they can
+				// assemble the full ChannelAnnouncement.
 				sigAnn, err := lnwire.NewAnnSigFromWireECDSARaw(
 					ann.ChannelID, ann.ShortChannelID,
-					remoteNSB, remoteBSB, nil,
+					localNSB, localBSB, nil,
 				)
-
 				if err != nil {
 					log.Errorf("Failed to generate "+
 						"announcement signature: %v",
 						err)
 					return
 				}
 
+				// Send our half of the proof to the peer. If
+				// successful, we mark that we've sent the proof
+				// during this connection so we don't re-send it
+				// on subsequent half-proof messages.
 				err = nMsg.peer.SendMessage(false, sigAnn)
 				if err != nil {
 					log.Errorf("Failed sending signature "+
@@ -3588,6 +3681,13 @@ func (d *AuthenticatedGossiper) handleAnnSig(ctx context.Context,
 					return
 				}
 
+				// With the announcement sent, we'll now mark
+				// that we've sent the proof to the peer for
+				// this connection.
+				d.sentAnnSigs.addToPeer(
+					peerVertex, ann.ChannelID,
+				)
+
 				log.Debugf("Signature announcement sent to "+
 					"peer=%x for chanID=%v", peerID,
 					ann.ChannelID)

discovery/gossiper_test.go

@@ -1783,16 +1783,23 @@ out:
 	}
 }
 
-// TestSignatureAnnouncementFullProofWhenRemoteProof tests that if a remote
+// TestSignatureAnnouncementResendWhenRemoteProof tests that if a remote
 // proof is received when we already have the full proof, the gossiper will send
-// the full proof (ChannelAnnouncement) to the remote peer.
-func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
+// our signature announcement max once per connection to the remote peer.
+func TestSignatureAnnouncementResendWhenRemoteProof(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
 	tCtx, err := createTestCtx(t, proofMatureDelta, false)
 	require.NoError(t, err, "can't create context")
 
+	// We'll create our test sync manager to have one active syncer.
+	syncMgr := newTestSyncManager(1)
+	syncMgr.Start()
+	defer syncMgr.Stop()
+
+	tCtx.gossiper.syncMgr = syncMgr
+
 	batch, err := tCtx.createLocalAnnouncements(0)
 	require.NoError(t, err, "can't generate announcements")
 
@@ -1806,8 +1813,15 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 		remoteKey, sentToPeer, tCtx.gossiper.quit, false,
 	)
 
+	// We create an active syncer for our remote peer.
+	tCtx.gossiper.InitSyncState(remotePeer)
+	remoteSyncer := assertSyncerExistence(t, syncMgr, remotePeer)
+	assertTransitionToChansSynced(t, remoteSyncer, remotePeer)
+	assertActiveGossipTimestampRange(t, remotePeer)
+	assertSyncerStatus(t, remoteSyncer, chansSynced, ActiveSync)
+
 	// Override NotifyWhenOnline to return the remote peer which we expect
-	// meesages to be sent to.
+	// messages to be sent to.
 	tCtx.gossiper.reliableSender.cfg.NotifyWhenOnline = func(_ [33]byte,
 		peerChan chan<- lnpeer.Peer) {
 
@@ -1905,14 +1919,25 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 	}
 	require.NoError(t, err, "unable to process local proof")
 
-	select {
-	case err = <-tCtx.gossiper.ProcessRemoteAnnouncement(
-		ctx, batch.remoteProofAnn, remotePeer,
-	):
-	case <-time.After(2 * time.Second):
-		t.Fatal("did not process local announcement")
+	// Define helper to process a remote proof in a sub-test.
+	processRemoteProof := func(t *testing.T) {
+		t.Helper()
+		var err error
+		select {
+		case err = <-tCtx.gossiper.ProcessRemoteAnnouncement(
+			ctx, batch.remoteProofAnn, remotePeer,
+		):
+		case <-time.After(2 * time.Second):
+			t.Fatal("did not process local announcement")
+		}
+		require.NoError(t, err, "unable to process remote proof")
 	}
-	require.NoError(t, err, "unable to process remote proof")
+
+	// Now give the gossiper the remote proof. This should
+	// trigger a send of our signature announcement.
+	t.Run("process remote proof - first send", func(t *testing.T) {
+		processRemoteProof(t)
+	})
 
 	// We expect the gossiper to send this message to the remote peer.
 	select {
@@ -1949,15 +1974,57 @@ func TestSignatureAnnouncementFullProofWhenRemoteProof(t *testing.T) {
 	}
 
 	// Now give the gossiper the remote proof yet again. This should
-	// trigger a send of the signature announcement.
+	// trigger a send of our signature announcement.
+	t.Run("process remote proof - first resend", func(t *testing.T) {
+		processRemoteProof(t)
+	})
+
+	// We expect the gossiper to send this message to the remote peer.
 	select {
-	case err = <-tCtx.gossiper.ProcessRemoteAnnouncement(
-		ctx, batch.remoteProofAnn, remotePeer,
-	):
+	case msg := <-sentToPeer:
+		_, ok := msg.(*lnwire.AnnounceSignatures1)
+		if !ok {
+			t.Fatalf("expected AnnounceSignatures1, instead got "+
+				"%T", msg)
+		}
 	case <-time.After(2 * time.Second):
-		t.Fatal("did not process local announcement")
+		t.Fatal("did not send local proof to peer")
 	}
-	require.NoError(t, err, "unable to process remote proof")
+
+	// Now give the gossiper the remote proof yet again. This should
+	// **not** trigger a send of our signature announcement.
+	t.Run("process remote proof - second resend", func(t *testing.T) {
+		processRemoteProof(t)
+	})
+
+	// We expect the gossiper to _not_ send this message to the remote peer.
+	select {
+	case msg := <-sentToPeer:
+		_, ok := msg.(*lnwire.AnnounceSignatures1)
+		if ok {
+			t.Fatalf("got an AnnounceSignatures1 when none was "+
+				"expected %T", msg)
+		}
+	case <-time.After(2 * time.Second):
+		break
+	}
+
+	// We prune the syncer, simulating the remote peer having disconnected.
+	tCtx.gossiper.PruneSyncState(remotePeer.PubKey())
+
+	// We simulate the remote peer coming back online.
+	tCtx.gossiper.InitSyncState(remotePeer)
+	remoteSyncer1 := assertSyncerExistence(t, syncMgr, remotePeer)
+	assertTransitionToChansSynced(t, remoteSyncer1, remotePeer)
+	assertActiveGossipTimestampRange(t, remotePeer)
+	assertSyncerStatus(t, remoteSyncer1, chansSynced, ActiveSync)
+
+	// Now give the gossiper the remote proof yet again. This should trigger
+	// a send of our signature announcement, because we are now on a new
+	// connection.
+	t.Run("process remote proof - third resend", func(t *testing.T) {
+		processRemoteProof(t)
+	})
 
 	// We expect the gossiper to send this message to the remote peer.
 	select {