feat(sharding)_: listen to new community shards together with old ones

Author: plopezlpz

messaging/api.go

@@ -234,8 +234,8 @@ func (a *API) ConnectionChanged(state connection.State) {
 	a.transport.ConnectionChanged(state)
 }
 
-func (a *API) SubscribeToPubsubTopic(topic string, optPublicKey *ecdsa.PublicKey) error {
-	return a.transport.SubscribeToPubsubTopic(topic, optPublicKey)
+func (a *API) SubscribeToPubsubTopic(topic string, optPublicKey ...*ecdsa.PublicKey) error {
+	return a.transport.SubscribeToPubsubTopic(topic, optPublicKey...)
 }
 
 func (a *API) UnsubscribeFromPubsubTopic(topic string) error {

messaging/transport/filters_manager.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"encoding/hex"
+	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -162,6 +164,8 @@ func (f *FiltersManager) InitCommunityFilters(communityFiltersToInitialize []Com
 
 		topics := make([]string, 0)
 		topics = append(topics, wakuv2.DefaultNonProtectedPubsubTopic())
+		topics = append(topics, wakuv2.GlobalCommunityControlPubsubTopic())
+		topics = append(topics, wakuv2.GlobalCommunityContentPubsubTopic())
 		topics = append(topics, communityFilter.Shard.PubsubTopic())
 
 		for _, pubsubTopic := range topics {
@@ -220,7 +224,16 @@ func (f *FiltersManager) Filters() (result []*Filter) {
 func (f *FiltersManager) Filter(chatID string) *Filter {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
-	return f.filters[chatID]
+
+	// find the first filter that matches this chat ID
+	// TODO this is temporary so not changing the return type, otherwise we should return a slice
+	for key, filter := range f.filters {
+		if strings.HasPrefix(key, chatID) {
+			return filter
+		}
+	}
+
+	return nil
 }
 
 // FilterByFilterID returns a Filter with a given Whisper filter ID.
@@ -276,19 +289,23 @@ func (f *FiltersManager) FilterByChatID(chatID string) *Filter {
 	return f.filters[chatID]
 }
 
-// Remove remove all the filters associated with a chat/identity
-func (f *FiltersManager) Remove(ctx context.Context, filters ...*Filter) error {
+// Remove removes all the filtersToRemove
+func (f *FiltersManager) Remove(ctx context.Context, filtersToRemove ...*Filter) error {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
 
-	for _, filter := range filters {
+	for _, filter := range filtersToRemove {
 		if err := f.service.Unsubscribe(ctx, filter.FilterID); err != nil {
 			return err
 		}
 		if filter.SymKeyID != "" {
 			f.service.DeleteSymKey(filter.SymKeyID)
 		}
-		delete(f.filters, filter.ChatID)
+		for k, v := range f.filters {
+			if filter.FilterID == v.FilterID {
+				delete(f.filters, k)
+			}
+		}
 	}
 
 	return nil
@@ -301,10 +318,10 @@ func (f *FiltersManager) RemoveNoListenFilters() error {
 	var filterIDs []string
 	var filters []*Filter
 
-	for _, f := range filters {
-		if !f.Listen {
-			filterIDs = append(filterIDs, f.FilterID)
-			filters = append(filters, f)
+	for _, v := range f.filters {
+		if !v.Listen {
+			filterIDs = append(filterIDs, v.FilterID)
+			filters = append(filters, v)
 		}
 	}
 	if err := f.service.UnsubscribeMany(filterIDs); err != nil {
@@ -315,30 +332,40 @@ func (f *FiltersManager) RemoveNoListenFilters() error {
 		if filter.SymKeyID != "" {
 			f.service.DeleteSymKey(filter.SymKeyID)
 		}
-		delete(f.filters, filter.ChatID)
+		for k, v := range f.filters {
+			if filter.FilterID == v.FilterID {
+				delete(f.filters, k)
+			}
+		}
 	}
 
 	return nil
 }
 
-// Remove remove all the filters associated with a chat/identity
+// RemoveFilterByChatID removes the filters associated with a chat/identity
 func (f *FiltersManager) RemoveFilterByChatID(chatID string) (*Filter, error) {
 	// TODO: remove subscriptions from waku2 if required. Might need to be implemented in transport
 
+	toRemove := make([]*Filter, 0)
 	f.mutex.Lock()
-	filter, ok := f.filters[chatID]
+	for _, filter := range f.filters {
+		if filter.ChatID == chatID {
+			toRemove = append(toRemove, filter)
+		}
+	}
 	f.mutex.Unlock()
 
-	if !ok {
+	if len(toRemove) == 0 {
 		return nil, nil
 	}
 
-	err := f.Remove(context.Background(), filter)
+	err := f.Remove(context.Background(), toRemove...)
 	if err != nil {
 		return nil, err
 	}
 
-	return filter, nil
+	// TODO temporary so not changing the return type, otherwise we should return a slice
+	return toRemove[0], nil
 }
 
 // LoadPartitioned creates a filter for a partitioned topic.
@@ -514,12 +541,14 @@ func (f *FiltersManager) PersonalTopicFilter() *Filter {
 	return f.filters[personalDiscoveryTopic]
 }
 
-// LoadPublic adds a filter for a public chat.
+// LoadPublic adds a filter for a public chat with specific pubsubTopic
 func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string) (*Filter, error) {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
 
-	if chat, ok := f.filters[chatID]; ok {
+	filterKey := toFilterKey(chatID, pubsubTopic)
+
+	if chat, ok := f.filters[filterKey]; ok {
 		if chat.PubsubTopic != pubsubTopic {
 			f.logger.Debug("updating pubsub topic for filter",
 				zap.String("chatID", chatID),
@@ -528,9 +557,8 @@ func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string) (*Filter,
 				zap.String("newTopic", pubsubTopic),
 			)
 			chat.PubsubTopic = pubsubTopic
-			f.filters[chatID] = chat
+			f.filters[filterKey] = chat
 		}
-
 		return chat, nil
 	}
 
@@ -550,7 +578,7 @@ func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string) (*Filter,
 		OneToOne:     false,
 	}
 
-	f.filters[chatID] = chat
+	f.filters[filterKey] = chat
 
 	f.logger.Debug("registering filter for",
 		zap.String("chatID", chatID),
@@ -681,3 +709,13 @@ func (f *FiltersManager) GetNegotiated(identity *ecdsa.PublicKey) *Filter {
 
 	return f.filters[NegotiatedTopic(identity)]
 }
+
+// toFilterKey creates a unique key for filters map using chatID and pubsubTopic
+//
+// to allow one chat to have multiple filters in different pubsubTopics so that we can migrate the communities to 128 and 256 shards
+func toFilterKey(chatID string, pubsubTopic string) string {
+	if pubsubTopic == "" {
+		return chatID
+	}
+	return fmt.Sprintf("%s::%s", chatID, pubsubTopic)
+}

messaging/transport/transport.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"database/sql"
+	"fmt"
 	"sync"
 	"time"
 
@@ -515,8 +516,25 @@ func (t *Transport) ConnectionChanged(state connection.State) {
 }
 
 // Subscribe to a pubsub topic, passing an optional public key if the pubsub topic is protected
-func (t *Transport) SubscribeToPubsubTopic(topic string, optPublicKey *ecdsa.PublicKey) error {
-	return t.waku.SubscribeToPubsubTopic(topic, optPublicKey)
+func (t *Transport) SubscribeToPubsubTopic(topic string, optPublicKeys ...*ecdsa.PublicKey) error {
+	if len(optPublicKeys) > 1 {
+		return fmt.Errorf("at most one public key can be passed, pub keys: %v", len(optPublicKeys))
+	}
+	var pubKey *ecdsa.PublicKey
+	if len(optPublicKeys) == 1 {
+		pubKey = optPublicKeys[0]
+	} else {
+		// try to retrieve pubkey for pubsubtopic if none provided
+		privK, err := t.RetrievePubsubTopicKey(topic)
+		if err != nil {
+			return err
+		}
+		if privK != nil {
+			pubKey = &privK.PublicKey
+		}
+	}
+	t.logger.Debug("subscribing to protected pubsub topic", zap.String("pubsubtopic", topic), zap.Any("pubkey", pubKey))
+	return t.waku.SubscribeToPubsubTopic(topic, pubKey)
 }
 
 // Unsubscribe from a pubsub topic

protocol/messenger_communities.go

@@ -1045,20 +1045,7 @@ func (m *Messenger) JoinCommunity(ctx context.Context, communityID types.HexByte
 }
 
 func (m *Messenger) subscribeToCommunityShard(communityID []byte, shard *wakuv2.Shard) error {
-	// TODO: this should probably be moved completely to transport once pubsub topic logic is implemented
-	pubsubTopic := shard.PubsubTopic()
-
-	privK, err := m.messaging.RetrievePubsubTopicKey(pubsubTopic)
-	if err != nil {
-		return err
-	}
-
-	var pubK *ecdsa.PublicKey
-	if privK != nil {
-		pubK = &privK.PublicKey
-	}
-
-	return m.messaging.SubscribeToPubsubTopic(pubsubTopic, pubK)
+	return m.messaging.SubscribeToPubsubTopic(shard.PubsubTopic())
 }
 
 func (m *Messenger) unsubscribeFromShard(shard *wakuv2.Shard) error {
@@ -2498,6 +2485,13 @@ func (m *Messenger) DefaultFilters(o *communities.Community) messaging.ChatsToIn
 		{ChatID: mlChannelID, PubsubTopic: communityPubsubTopic},
 		{ChatID: memberUpdateChannelID, PubsubTopic: communityPubsubTopic},
 		{ChatID: uncompressedPubKey, PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic()},
+
+		// While we migrate to 128 and 256 shards, we listen to community messages in the new shards as well as in the old ones. After migration we should remove the old ones
+		{ChatID: cID, PubsubTopic: wakuv2.GlobalCommunityControlPubsubTopic()},
+		{ChatID: updatesChannelID, PubsubTopic: wakuv2.GlobalCommunityControlPubsubTopic()},
+		{ChatID: mlChannelID, PubsubTopic: wakuv2.GlobalCommunityControlPubsubTopic()},
+		{ChatID: memberUpdateChannelID, PubsubTopic: wakuv2.GlobalCommunityContentPubsubTopic()}, // Making content since chat messages are sent in this contenttopic
+		{ChatID: uncompressedPubKey, PubsubTopic: wakuv2.GlobalCommunityContentPubsubTopic()},
 	}
 
 	return chats

protocol/messenger_filter_init.go

@@ -24,7 +24,16 @@ func (m *Messenger) InitFilters() error {
 	rand.Seed(time.Now().Unix())
 
 	// Community requests will arrive in this pubsub topic
-	if err := m.SubscribeToPubsubTopic(wakuv2.DefaultNonProtectedPubsubTopic(), nil); err != nil {
+	// TODO depracate
+	if err := m.SubscribeToPubsubTopic(wakuv2.DefaultNonProtectedPubsubTopic()); err != nil {
+		return err
+	}
+	// TODO only subscribe if interested in communities
+	if err := m.SubscribeToPubsubTopic(wakuv2.GlobalCommunityControlPubsubTopic()); err != nil {
+		return err
+	}
+	// TODO only subscribe if interested in communities
+	if err := m.SubscribeToPubsubTopic(wakuv2.GlobalCommunityContentPubsubTopic()); err != nil {
 		return err
 	}
 

protocol/messenger_peers.go

@@ -44,8 +44,8 @@ func (m *Messenger) ENR() (*enode.Node, error) {
 }
 
 // Subscribe to a pubsub topic, passing an optional public key if the pubsub topic is protected
-func (m *Messenger) SubscribeToPubsubTopic(topic string, optPublicKey *ecdsa.PublicKey) error {
-	return m.messaging.SubscribeToPubsubTopic(topic, optPublicKey)
+func (m *Messenger) SubscribeToPubsubTopic(topic string, optPublicKey ...*ecdsa.PublicKey) error {
+	return m.messaging.SubscribeToPubsubTopic(topic, optPublicKey...)
 }
 
 func (m *Messenger) StorePubsubTopicKey(topic string, privKey *ecdsa.PrivateKey) error {

protocol/messenger_store_node_request_manager.go

@@ -86,46 +86,72 @@ func (m *StoreNodeRequestManager) FetchCommunity(ctx context.Context, community
 		zap.Any("community", community),
 		zap.Any("config", cfg))
 
-	requestCommunity := func(communityID string, shard *wakuv2.Shard) (*communities.Community, StoreNodeRequestStats, error) {
-		channel, err := m.subscribeToRequest(ctx, storeNodeCommunityRequest, communityID, shard, cfg)
-		if err != nil {
-			return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
+	fetch := func() (*communities.Community, StoreNodeRequestStats, error) {
+		// if shard was not passed or nil, see if this community has a custom shard by requesting for this config to the store node
+		communityCustomShard := community.Shard
+		if communityCustomShard == nil {
+			var err error
+			communityCustomShard, err = m.fetchCustomShard(ctx, community.CommunityID, cfg, wakuv2.DefaultNonProtectedShard(), wakuv2.GlobalCommunityControlShard())
+			if err != nil {
+				return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to fetch community shard: %w", err)
+			}
 		}
-
-		if !cfg.WaitForResponse {
-			return nil, StoreNodeRequestStats{}, nil
+		if communityCustomShard != nil {
+			return m.fetchCommunity(ctx, community.CommunityID, cfg, communityCustomShard)
 		}
+		// request community in 32 and 64 shards
+		return m.fetchCommunity(ctx, community.CommunityID, cfg, wakuv2.DefaultShard(), wakuv2.GlobalCommunityControlShard())
+	}
 
-		result := <-channel
-		return result.community, result.stats, result.err
+	if !cfg.WaitForResponse {
+		go fetch()
+		return nil, StoreNodeRequestStats{}, nil
 	}
+	return fetch()
+}
 
-	// if shard was not passed or nil, request shard first
-	communityShard := community.Shard
-	if communityShard == nil {
-		id := messaging.CommunityShardInfoTopic(community.CommunityID)
-		fetchedShard, err := m.subscribeToRequest(ctx, storeNodeShardRequest, id, wakuv2.DefaultNonProtectedShard(), cfg)
+// fetchCommunity fetches a community from the store node.
+//
+// Since now we could have this community description in shards 32 and 128 we are doing the following:
+//  1. trying to fetch in shards 32 and 128
+//  2. returning the first non-nil community result
+//
+// Eventually we should just go to shard 128, once full migration to Global Community Control and Content Topic is done.
+func (m *StoreNodeRequestManager) fetchCommunity(ctx context.Context, communityID string, cfg StoreNodeRequestConfig, shards ...*wakuv2.Shard) (*communities.Community, StoreNodeRequestStats, error) {
+	for _, shard := range shards {
+		sub, err := m.subscribeToRequest(ctx, storeNodeCommunityRequest, communityID, shard, cfg)
 		if err != nil {
 			return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err)
 		}
-
-		if !cfg.WaitForResponse {
-			go func() {
-				defer gocommon.LogOnPanic()
-				shardResult := <-fetchedShard
-				communityShard = shardResult.shard
-
-				_, _, _ = requestCommunity(community.CommunityID, communityShard)
-			}()
-			return nil, StoreNodeRequestStats{}, nil
+		// not doing this concurrently to avoid two filters for the same communityID, not a big deal since it is at most 2 shards and temporary
+		res := <-sub
+		if res.community != nil {
+			return res.community, res.stats, res.err
 		}
-
-		shardResult := <-fetchedShard
-		communityShard = shardResult.shard
 	}
+	return nil, StoreNodeRequestStats{}, nil
+}
 
-	// request community with on shard
-	return requestCommunity(community.CommunityID, communityShard)
+// fetchCustomShard fetches a community custom shard from the store node.
+//
+// Since now we could have communities' shard info in shards 64 and 128 we are doing the following:
+//  1. trying to fetch in shards 64 and 128
+//  2. returning the first non-nil shard result
+//
+// Eventually we should just go to shard 128, once full migration to Global Community Control and Content Topic is done.
+func (m *StoreNodeRequestManager) fetchCustomShard(ctx context.Context, communityID string, cfg StoreNodeRequestConfig, shards ...*wakuv2.Shard) (*wakuv2.Shard, error) {
+	for _, shard := range shards {
+		sub, err := m.subscribeToRequest(ctx, storeNodeCommunityRequest, messaging.CommunityShardInfoTopic(communityID), shard, cfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create a shard info request: %w", err)
+		}
+		// not doing this concurrently to avoid two filters for the same CommunityShardInfoTopic, not a big deal since it is at most 2 shards and temporary
+		res := <-sub
+		if res.community != nil {
+			return shard, nil
+		}
+	}
+	return nil, nil
 }
 
 // FetchCommunities makes a FetchCommunity for each element in given `communities` list.