txnprovider/shutter: fix premature encrypted txn pool cleanup bug and peer drops (#18264)

for #18217 and
#18263

Author: taratorio

txnprovider/shutter/decryption_keys_processor.go

@@ -345,7 +345,7 @@ func (dkp *DecryptionKeysProcessor) processBlockEventCleanup(blockEvent BlockEve
 
 	for _, mark := range cleanUpMarks {
 		dkp.processed.Remove(mark)
-		dkp.encryptedTxnsPool.DeleteUpTo(mark.Eon, mark.To+1)
+		dkp.encryptedTxnsPool.DeleteUpTo(mark.Eon, mark.To)
 	}
 
 	return nil

txnprovider/shutter/decryption_keys_source.go

@@ -29,6 +29,7 @@ import (
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	libp2pcrypto "github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/multiformats/go-multiaddr"
 	"golang.org/x/sync/errgroup"
@@ -136,10 +137,11 @@ func (dks *PubSubDecryptionKeysSource) Run(ctx context.Context) error {
 	})
 
 	eg.Go(func() error {
-		select {
-		case <-ctx.Done(): // to keep the host and topic alive until Run's ctx is cancelled
-			return ctx.Err()
+		err := dks.reconnectBootstrapNodes(ctx, p2pHost)
+		if err != nil {
+			return fmt.Errorf("reconnect bootstrap nodes failure: %w", err)
 		}
+		return nil
 	})
 
 	return eg.Wait()
@@ -194,6 +196,7 @@ func (dks *PubSubDecryptionKeysSource) initP2pHost() (host.Host, error) {
 		return nil, err
 	}
 
+	p2pHost.Network().Notify(&peerNotifiee{logger: dks.logger})
 	dks.logger.Info("shutter p2p host initialised", "addr", listenAddr, "id", p2pHost.ID())
 	return p2pHost, nil
 }
@@ -331,6 +334,34 @@ func (dks *PubSubDecryptionKeysSource) peerInfoLoop(ctx context.Context, pubSub
 	}
 }
 
+func (dks *PubSubDecryptionKeysSource) reconnectBootstrapNodes(ctx context.Context, host host.Host) error {
+	bootstrapNodes, err := dks.config.BootstrapNodesAddrInfo()
+	if err != nil {
+		return err
+	}
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			for _, node := range bootstrapNodes {
+				connectedness := host.Network().Connectedness(node.ID)
+				if connectedness != network.Connected {
+					dks.logger.Warn("lost connection to shutter bootstrap node, reconnecting", "node", node, "connectedness", connectedness)
+					err := host.Connect(ctx, node)
+					if err != nil {
+						dks.logger.Warn("failed to reconnect to bootstrap node", "node", node, "err", err)
+					} else {
+						dks.logger.Info("reconnected to shutter bootstrap node", "node", node)
+					}
+				}
+			}
+		}
+	}
+}
+
 func decryptionKeysTopicScoreParams() *pubsub.TopicScoreParams {
 	// NOTE: this is taken from
 	// https://github.com/shutter-network/rolling-shutter/blob/main/rolling-shutter/p2p/params.go#L100
@@ -358,3 +389,23 @@ func decryptionKeysTopicScoreParams() *pubsub.TopicScoreParams {
 		InvalidMessageDeliveriesDecay:   0.9994,
 	}
 }
+
+type peerNotifiee struct {
+	logger log.Logger
+}
+
+func (n *peerNotifiee) Listen(network.Network, multiaddr.Multiaddr) {
+	// no-op
+}
+
+func (n *peerNotifiee) ListenClose(network.Network, multiaddr.Multiaddr) {
+	// no-op
+}
+
+func (n *peerNotifiee) Connected(_ network.Network, conn network.Conn) {
+	n.logger.Debug("shutter peer connected", "peer", conn.RemotePeer(), "addr", conn.RemoteMultiaddr())
+}
+
+func (n *peerNotifiee) Disconnected(_ network.Network, conn network.Conn) {
+	n.logger.Debug("shutter peer disconnected", "peer", conn.RemotePeer(), "addr", conn.RemoteMultiaddr())
+}

txnprovider/shutter/pool_test.go

@@ -69,10 +69,12 @@ func TestPoolCleanup(t *testing.T) {
 		// simulate the first block
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 0)
+		require.Len(t, pool.AllDecryptedTxns(), 0)
 		// simulate some encrypted txn submissions and simulate a new block
 		encTxn1 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
 		encTxn2 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
-		require.Len(t, pool.AllEncryptedTxns(), 0)
 		err = handle.SimulateLogEvents(ctx, []types.Log{
 			MockTxnSubmittedEventLog(t, handle.config, ekg.Eon(), 1, encTxn1),
 			MockTxnSubmittedEventLog(t, handle.config, ekg.Eon(), 2, encTxn2),
@@ -83,29 +85,114 @@ func TestPoolCleanup(t *testing.T) {
 		require.NoError(t, err)
 		synctest.Wait()
 		require.Len(t, pool.AllEncryptedTxns(), 2)
+		require.Len(t, pool.AllDecryptedTxns(), 0)
 		// simulate decryption keys
 		handle.SimulateCurrentSlot()
 		handle.SimulateDecryptionKeys(ctx, t, ekg, 1, encTxn1.IdentityPreimage, encTxn2.IdentityPreimage)
 		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 2)
 		require.Len(t, pool.AllDecryptedTxns(), 2)
 		// simulate one block passing by - decrypted txns pool should get cleaned up after 1 slot
 		handle.SimulateCachedEonRead(t, ekg)
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
 		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 2)
 		require.Len(t, pool.AllDecryptedTxns(), 0)
 		// simulate more blocks passing by - encrypted txns pool should get cleaned up after config.ReorgDepthAwareness
 		handle.SimulateCachedEonRead(t, ekg)
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 2)
+		require.Len(t, pool.AllDecryptedTxns(), 0)
+		handle.SimulateCachedEonRead(t, ekg)
+		err = handle.SimulateNewBlockChange(ctx)
+		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 2)
+		require.Len(t, pool.AllDecryptedTxns(), 0)
+		handle.SimulateCachedEonRead(t, ekg)
+		err = handle.SimulateNewBlockChange(ctx)
+		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 0)
+		require.Len(t, pool.AllDecryptedTxns(), 0)
+	})
+}
+
+func TestPoolCleanupShouldNotDeleteNewEncTxnsDueToConsecutiveEmptyDecrMsgs(t *testing.T) {
+	/*
+		Reproduces the following bug:
+			1. we were getting decryption key messages for our slots with 0 keys for transactions (valid behaviour when no transactions to decrypt)
+			2. we got an encrypted txn submission at block 19,182,545
+			3. got block event for 19,182,546 (next block)
+			4. encrypted txn got deleted from the pool (although it was just added 1 block ago)
+			5. the decryption keys arrived for the slot of block 19,182,547 and the txn was missed
+
+		[DBUG] [12-10|13:05:25.131] received encrypted txn submission event		component=shutter eonIndex=14 txnIndex=47 blockNum=19182545 unwind=false
+		...
+		[DBUG] [12-10|13:05:30.155] block tracker got block event				component=shutter blockNum=19182546
+		[DBUG] [12-10|13:05:30.155] deleted encrypted txns						component=shutter count=1 fromEon=14 fromTxn=47 toEon=14 toTxn=47
+		...
+		[DBUG] [12-10|13:05:30.342] processing decryption keys message			component=shutter instanceId=102000 eonIndex=14 slot=19995127 txnIndex=47 keys=2
+		[DBUG] [12-10|13:05:30.342] looking up encrypted txns for 				component=shutter eon=14 from=47 to=48 gasLimit=10000000
+	*/
+	t.Parallel()
+	pt := PoolTest{t}
+	pt.Run(func(ctx context.Context, t *testing.T, pool *shutter.Pool, handle PoolTestHandle) {
+		// simulate expected contract calls for reading the first ekg after the first block event
+		ekg, err := testhelpers.MockEonKeyGeneration(shutter.EonIndex(0), 1, 2, 1)
+		require.NoError(t, err)
+		handle.SimulateInitialEonRead(t, ekg)
+		// simulate loadSubmissions after the first block
+		handle.SimulateFilterLogs(common.HexToAddress(handle.config.SequencerContractAddress), []types.Log{})
+		// simulate the first block
+		err = handle.SimulateNewBlockChange(ctx)
+		require.NoError(t, err)
+		synctest.Wait()
+		// simulate decryption keys msg with 0 decrypted txns for 2 slots in a row to get close to the currently
+		// configured reorg awareness window of 3 (make sure that is the case otherwise the test won't be accurately
+		// capturing the situation)
+		require.Equal(t, uint64(3), handle.config.ReorgDepthAwareness)
+		// 1
+		handle.SimulateCurrentSlot()
+		handle.SimulateDecryptionKeys(ctx, t, ekg, 1)
 		handle.SimulateCachedEonRead(t, ekg)
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 0)
+		// 2
+		handle.SimulateCurrentSlot()
+		handle.SimulateDecryptionKeys(ctx, t, ekg, 1)
 		handle.SimulateCachedEonRead(t, ekg)
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
 		synctest.Wait()
 		require.Len(t, pool.AllEncryptedTxns(), 0)
+		// simulate a new encrypted txn submission
+		// 3
+		encTxn1 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
+		err = handle.SimulateLogEvents(ctx, []types.Log{
+			MockTxnSubmittedEventLog(t, handle.config, ekg.Eon(), 1, encTxn1),
+		})
+		require.NoError(t, err)
+		handle.SimulateCachedEonRead(t, ekg)
+		err = handle.SimulateNewBlockChange(ctx)
+		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 1)
+		// simulate decryption keys msg with 0 decrypted txns for the next slot
+		// the encrypted txn should not get deleted!
+		// 4
+		handle.SimulateCurrentSlot()
+		handle.SimulateDecryptionKeys(ctx, t, ekg, 1)
+		handle.SimulateCachedEonRead(t, ekg)
+		err = handle.SimulateNewBlockChange(ctx)
+		require.NoError(t, err)
+		synctest.Wait()
+		require.Len(t, pool.AllEncryptedTxns(), 1)
 	})
 }
 
@@ -122,6 +209,7 @@ func TestPoolSkipsBlobTxns(t *testing.T) {
 		// simulate the first block
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
+		synctest.Wait()
 		// simulate some encrypted txn submissions and simulate a new block
 		encBlobTxn1 := MockEncryptedBlobTxn(t, handle.config.ChainId, ekg.Eon())
 		encTxn1 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
@@ -167,6 +255,7 @@ func TestPoolProvideTxnsUsesGasTargetAndTxnsIdFilter(t *testing.T) {
 		// simulate the first block
 		err = handle.SimulateNewBlockChange(ctx)
 		require.NoError(t, err)
+		synctest.Wait()
 		// simulate some encrypted txn submissions and simulate a new block
 		encTxn1 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
 		encTxn2 := MockEncryptedTxn(t, handle.config.ChainId, ekg.Eon())
@@ -220,7 +309,7 @@ func (t PoolTest) Run(testCase func(ctx context.Context, t *testing.T, pool *shu
 	synctest.Test(t.T, func(t *testing.T) {
 		ctx, cancel := context.WithCancel(t.Context())
 		defer cancel()
-		logger := testlog.Logger(t, log.LvlTrace)
+		logger := testlog.Logger(t, log.LvlCrit)
 		logHandler := testhelpers.NewCollectingLogHandler(logger.GetHandler())
 		logger.SetHandler(logHandler)
 		config := shuttercfg.ConfigByChainName(networkname.Chiado)