Integrate end to end hashed chain exchange with F3 (#816)

Integrate the chain exchange mechanism with F3 host and runner. But
without touching the core GPBFT.

The implementation here leaves two major TODOs: 1) chain broadcasting
mechanism (currently coupled to GPBFT message broadcast), and 2)
partial message validation prior to buffering (currently skipped
entirely but with capped buffer sizes and re-validation by core GPBFT
once the messages are complete).

The integration introduces the concept of Partial GMessage: a GMessage
with chains replaced with the key to the chain. The work introduces a
buffer and refill mechanism that listens to the chains discovered,
un-buffers the messages having re-constructed their original GMessage
and feeds them to the participation using the existing event loop.

Part of #792

Author: masih

Diff for: cbor_gen.go

@@ -32,11 +32,12 @@ type PubSubChainExchange struct {
 	*options
 
 	// mu guards access to chains and API calls.
-	mu               sync.Mutex
-	chainsWanted     map[uint64]*lru.Cache[string, *chainPortion]
-	chainsDiscovered map[uint64]*lru.Cache[string, *chainPortion]
-	topic            *pubsub.Topic
-	stop             func() error
+	mu                   sync.Mutex
+	chainsWanted         map[uint64]*lru.Cache[string, *chainPortion]
+	chainsDiscovered     map[uint64]*lru.Cache[string, *chainPortion]
+	pendingCacheAsWanted chan Message
+	topic                *pubsub.Topic
+	stop                 func() error
 }
 
 func NewPubSubChainExchange(o ...Option) (*PubSubChainExchange, error) {
@@ -45,9 +46,10 @@ func NewPubSubChainExchange(o ...Option) (*PubSubChainExchange, error) {
 		return nil, err
 	}
 	return &PubSubChainExchange{
-		options:          opts,
-		chainsWanted:     map[uint64]*lru.Cache[string, *chainPortion]{},
-		chainsDiscovered: map[uint64]*lru.Cache[string, *chainPortion]{},
+		options:              opts,
+		chainsWanted:         map[uint64]*lru.Cache[string, *chainPortion]{},
+		chainsDiscovered:     map[uint64]*lru.Cache[string, *chainPortion]{},
+		pendingCacheAsWanted: make(chan Message, 100), // TODO: parameterise.
 	}, nil
 }
 
@@ -64,7 +66,9 @@ func (p *PubSubChainExchange) Start(ctx context.Context) error {
 	}
 	if p.topicScoreParams != nil {
 		if err := p.topic.SetScoreParams(p.topicScoreParams); err != nil {
-			return fmt.Errorf("failed to set score params: %w", err)
+			// This can happen most likely due to router not supporting peer scoring. It's
+			// non-critical. Hence, the warning log.
+			log.Warnw("failed to set topic score params", "err", err)
 		}
 	}
 	subscription, err := p.topic.Subscribe(pubsub.WithBufferSize(p.subscriptionBufferSize))
@@ -79,17 +83,31 @@ func (p *PubSubChainExchange) Start(ctx context.Context) error {
 		for ctx.Err() == nil {
 			msg, err := subscription.Next(ctx)
 			if err != nil {
-				log.Debugw("failed to read nex message from subscription", "err", err)
+				log.Debugw("failed to read next message from subscription", "err", err)
 				continue
 			}
 			cmsg := msg.ValidatorData.(Message)
 			p.cacheAsDiscoveredChain(ctx, cmsg)
 		}
+		log.Debug("Stopped reading messages from chainexchange subscription.")
+	}()
+	go func() {
+		for ctx.Err() == nil {
+			select {
+			case <-ctx.Done():
+				return
+			case cmsg := <-p.pendingCacheAsWanted:
+				p.cacheAsWantedChain(ctx, cmsg)
+			}
+		}
+		log.Debug("Stopped caching chains as wanted.")
 	}()
 	p.stop = func() error {
 		cancel()
 		subscription.Cancel()
-		return p.topic.Close()
+		_ = p.pubsub.UnregisterTopicValidator(p.topicName)
+		_ = p.topic.Close()
+		return nil
 	}
 	return nil
 }
@@ -124,21 +142,18 @@ func (p *PubSubChainExchange) GetChainByInstance(ctx context.Context, instance u
 	cacheKey := string(key)
 
 	// Check wanted keys first.
-	p.mu.Lock()
+
 	wanted := p.getChainsWantedAt(instance)
-	p.mu.Unlock()
 	if portion, found := wanted.Get(cacheKey); found && !portion.IsPlaceholder() {
 		return portion.chain, true
 	}
 
 	// Check if the chain for the key is discovered.
-	p.mu.Lock()
 	discovered := p.getChainsDiscoveredAt(instance)
 	if portion, found := discovered.Get(cacheKey); found {
 		// Add it to the wanted cache and remove it from the discovered cache.
 		wanted.Add(cacheKey, portion)
 		discovered.Remove(cacheKey)
-		p.mu.Unlock()
 
 		chain := portion.chain
 		if p.listener != nil {
@@ -147,7 +162,6 @@ func (p *PubSubChainExchange) GetChainByInstance(ctx context.Context, instance u
 		// TODO: Do we want to pull all the suffixes of the chain into wanted cache?
 		return chain, true
 	}
-	p.mu.Unlock()
 
 	// Otherwise, add a placeholder for the wanted key as a way to prioritise its
 	// retention via LRU recent-ness.
@@ -156,6 +170,8 @@ func (p *PubSubChainExchange) GetChainByInstance(ctx context.Context, instance u
 }
 
 func (p *PubSubChainExchange) getChainsWantedAt(instance uint64) *lru.Cache[string, *chainPortion] {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	wanted, exists := p.chainsWanted[instance]
 	if !exists {
 		wanted = p.newChainPortionCache(p.maxWantedChainsPerInstance)
@@ -165,6 +181,8 @@ func (p *PubSubChainExchange) getChainsWantedAt(instance uint64) *lru.Cache[stri
 }
 
 func (p *PubSubChainExchange) getChainsDiscoveredAt(instance uint64) *lru.Cache[string, *chainPortion] {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	discovered, exists := p.chainsDiscovered[instance]
 	if !exists {
 		discovered = p.newChainPortionCache(p.maxDiscoveredChainsPerInstance)
@@ -208,8 +226,6 @@ func (p *PubSubChainExchange) validatePubSubMessage(_ context.Context, _ peer.ID
 }
 
 func (p *PubSubChainExchange) cacheAsDiscoveredChain(ctx context.Context, cmsg Message) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
 
 	wanted := p.getChainsDiscoveredAt(cmsg.Instance)
 	discovered := p.getChainsDiscoveredAt(cmsg.Instance)
@@ -245,7 +261,13 @@ func (p *PubSubChainExchange) cacheAsDiscoveredChain(ctx context.Context, cmsg M
 func (p *PubSubChainExchange) Broadcast(ctx context.Context, msg Message) error {
 
 	// Optimistically cache the broadcast chain and all of its prefixes as wanted.
-	p.cacheAsWantedChain(ctx, msg)
+	select {
+	case p.pendingCacheAsWanted <- msg:
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		log.Warnw("Dropping wanted cache entry. Chain exchange is too slow to process chains as wanted", "msg", msg)
+	}
 
 	// TODO: integrate zstd compression.
 	var buf bytes.Buffer
@@ -266,7 +288,6 @@ type discovery struct {
 
 func (p *PubSubChainExchange) cacheAsWantedChain(ctx context.Context, cmsg Message) {
 	var notifications []discovery
-	p.mu.Lock()
 	wanted := p.getChainsWantedAt(cmsg.Instance)
 	for offset := len(cmsg.Chain); offset >= 0 && ctx.Err() == nil; offset-- {
 		// TODO: Expose internals of merkle.go so that keys can be generated
@@ -290,7 +311,6 @@ func (p *PubSubChainExchange) cacheAsWantedChain(ctx context.Context, cmsg Messa
 		// been evicted from the cache or not. This should be cheap enough considering the
 		// added complexity of tracking evictions relative to chain prefixes.
 	}
-	p.mu.Unlock()
 
 	// Notify the listener outside the lock.
 	if p.listener != nil {

Diff for: chainexchange/pubsub.go

@@ -2,6 +2,8 @@ package chainexchange_test
 
 import (
 	"context"
+	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -52,32 +54,38 @@ func TestPubSubChainExchange_Broadcast(t *testing.T) {
 	chain, found := subject.GetChainByInstance(ctx, instance, key)
 	require.False(t, found)
 	require.Nil(t, chain)
-	require.Empty(t, testListener.notifications)
+	require.Empty(t, testListener.getNotifications())
 
 	require.NoError(t, subject.Broadcast(ctx, chainexchange.Message{
 		Instance: instance,
 		Chain:    ecChain,
 	}))
 
-	chain, found = subject.GetChainByInstance(ctx, instance, key)
-	require.True(t, found)
+	require.Eventually(t, func() bool {
+		chain, found = subject.GetChainByInstance(ctx, instance, key)
+		return found
+	}, time.Second, 100*time.Millisecond)
 	require.Equal(t, ecChain, chain)
 
 	baseChain := ecChain.BaseChain()
 	baseKey := subject.Key(baseChain)
-	chain, found = subject.GetChainByInstance(ctx, instance, baseKey)
-	require.True(t, found)
+	require.Eventually(t, func() bool {
+		chain, found = subject.GetChainByInstance(ctx, instance, baseKey)
+		return found
+	}, time.Second, 100*time.Millisecond)
 	require.Equal(t, baseChain, chain)
 
 	// Assert that we have received 2 notifications, because ecChain has 2 tipsets.
 	// First should be the ecChain, second should be the baseChain.
-	require.Len(t, testListener.notifications, 2)
-	require.Equal(t, instance, testListener.notifications[1].instance)
-	require.Equal(t, baseKey, testListener.notifications[1].key)
-	require.Equal(t, baseChain, testListener.notifications[1].chain)
-	require.Equal(t, instance, testListener.notifications[0].instance)
-	require.Equal(t, key, testListener.notifications[0].key)
-	require.Equal(t, ecChain, testListener.notifications[0].chain)
+
+	notifications := testListener.getNotifications()
+	require.Len(t, notifications, 2)
+	require.Equal(t, instance, notifications[1].instance)
+	require.Equal(t, baseKey, notifications[1].key)
+	require.Equal(t, baseChain, notifications[1].chain)
+	require.Equal(t, instance, notifications[0].instance)
+	require.Equal(t, key, notifications[0].key)
+	require.Equal(t, ecChain, notifications[0].chain)
 
 	require.NoError(t, subject.Shutdown(ctx))
 }
@@ -88,13 +96,22 @@ type notification struct {
 	chain    gpbft.ECChain
 }
 type listener struct {
+	mu            sync.Mutex
 	notifications []notification
 }
 
 func (l *listener) NotifyChainDiscovered(_ context.Context, key chainexchange.Key, instance uint64, chain gpbft.ECChain) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
 	l.notifications = append(l.notifications, notification{key: key, instance: instance, chain: chain})
 }
 
+func (l *listener) getNotifications() []notification {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return slices.Clone(l.notifications)
+}
+
 // TODO: Add more tests, specifically:
 //        - validation
 //        - discovery through other chainexchange instance

Diff for: chainexchange/pubsub_test.go

@@ -548,7 +548,7 @@ func (e *testEnv) waitForEpochFinalized(epoch int64) {
 					}
 				}
 				return false
-			}, 30*time.Second)
+			}, 60*time.Second)
 		}
 
 		if head < epoch-100 {