mirror of
https://github.com/status-im/go-waku.git
synced 2025-01-12 14:54:19 +00:00
242 lines
9.1 KiB
Go
242 lines
9.1 KiB
Go
//go:build include_storev3_tests
|
|
// +build include_storev3_tests
|
|
|
|
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/crypto"
|
|
"github.com/ethereum/go-ethereum/p2p/enode"
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
|
"github.com/multiformats/go-multiaddr"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/waku-org/go-waku/tests"
|
|
"github.com/waku-org/go-waku/waku/v2/peermanager"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol/enr"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol/metadata"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
|
|
"github.com/waku-org/go-waku/waku/v2/timesource"
|
|
"github.com/waku-org/go-waku/waku/v2/utils"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
func TestStoreClient(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
defer cancel()
|
|
|
|
port, err := tests.FindFreePort(t, "", 5)
|
|
require.NoError(t, err)
|
|
|
|
host, err := tests.MakeHost(context.Background(), port, rand.Reader)
|
|
require.NoError(t, err)
|
|
|
|
db, err := enode.OpenDB("")
|
|
require.NoError(t, err)
|
|
priv, err := crypto.GenerateKey()
|
|
require.NoError(t, err)
|
|
localnode := enode.NewLocalNode(db, priv)
|
|
|
|
pubsubTopic := "/waku/2/rs/99/1"
|
|
clusterID := uint16(99)
|
|
rs, _ := protocol.NewRelayShards(clusterID, 1)
|
|
enr.Update(utils.Logger(), localnode, enr.WithWakuRelaySharding(rs))
|
|
|
|
metadata := metadata.NewWakuMetadata(clusterID, localnode, utils.Logger())
|
|
metadata.SetHost(host)
|
|
metadata.Start(context.Background())
|
|
|
|
// Creating a relay instance for pushing messages to the store node
|
|
b := relay.NewBroadcaster(10)
|
|
require.NoError(t, b.Start(context.Background()))
|
|
wakuRelay := relay.NewWakuRelay(b, 0, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, utils.Logger())
|
|
wakuRelay.SetHost(host)
|
|
require.NoError(t, err)
|
|
err = wakuRelay.Start(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
pm := peermanager.NewPeerManager(5, 5, nil, wakuRelay, true, utils.Logger())
|
|
pm.SetHost(host)
|
|
err = pm.SubscribeToRelayEvtBus(wakuRelay.Events())
|
|
require.NoError(t, err)
|
|
pm.Start(ctx)
|
|
|
|
// Creating a storeV3 instance for all queries
|
|
wakuStore := NewWakuStore(pm, timesource.NewDefaultClock(), utils.Logger(), 8)
|
|
wakuStore.SetHost(host)
|
|
|
|
_, err = wakuRelay.Subscribe(context.Background(), protocol.NewContentFilter(pubsubTopic), relay.WithoutConsumer())
|
|
require.NoError(t, err)
|
|
|
|
// Obtain multiaddr from env
|
|
envStorenode := os.Getenv("TEST_STOREV3_NODE")
|
|
if envStorenode == "" {
|
|
envStorenode = "/ip4/127.0.0.1/tcp/60000/p2p/16Uiu2HAmMGhfSTUzKbsjMWxc6T1X4wiTWSF1bEWSLjAukCm7KiHV"
|
|
}
|
|
storenode_multiaddr, err := multiaddr.NewMultiaddr(envStorenode)
|
|
require.NoError(t, err)
|
|
|
|
storenode, err := peer.AddrInfoFromP2pAddr(storenode_multiaddr)
|
|
require.NoError(t, err)
|
|
|
|
err = host.Connect(ctx, *storenode)
|
|
require.NoError(t, err)
|
|
|
|
// Wait until mesh forms
|
|
time.Sleep(2 * time.Second)
|
|
|
|
// Send messages
|
|
messages := []*pb.WakuMessage{}
|
|
startTime := utils.GetUnixEpoch(timesource.NewDefaultClock())
|
|
for i := 0; i < 5; i++ {
|
|
msg := &pb.WakuMessage{
|
|
Payload: []byte{byte(i), 1, 2, 3, 4, 5},
|
|
ContentTopic: "test",
|
|
Version: proto.Uint32(0),
|
|
Timestamp: utils.GetUnixEpoch(timesource.NewDefaultClock()),
|
|
}
|
|
_, err := wakuRelay.Publish(ctx, msg, relay.WithPubSubTopic(pubsubTopic))
|
|
require.NoError(t, err)
|
|
|
|
messages = append(messages, msg)
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
endTime := utils.GetUnixEpoch(timesource.NewDefaultClock())
|
|
|
|
time.Sleep(1 * time.Second)
|
|
|
|
// Check for message existence
|
|
exists, err := wakuStore.Exists(ctx, messages[0].Hash(pubsubTopic), WithPeer(storenode.ID))
|
|
require.NoError(t, err)
|
|
require.True(t, exists)
|
|
|
|
// Message should not exist
|
|
exists, err = wakuStore.Exists(ctx, pb.MessageHash{1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2}, WithPeer(storenode.ID))
|
|
require.NoError(t, err)
|
|
require.False(t, exists)
|
|
|
|
// Query messages with forward pagination
|
|
response, err := wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(true, 2))
|
|
require.NoError(t, err)
|
|
|
|
// -- First page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 2)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[0].GetTimestamp())
|
|
require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[1].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Second page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 2)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[2].GetTimestamp())
|
|
require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[3].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Third page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 1)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[4].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Trying to continue a completed cursor
|
|
require.True(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 0)
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// Query messages with backward pagination
|
|
response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(false, 2))
|
|
require.NoError(t, err)
|
|
|
|
// -- First page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 2)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[3].GetTimestamp())
|
|
require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[4].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Second page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 2)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[1].GetTimestamp())
|
|
require.Equal(t, response.Messages()[1].Message.GetTimestamp(), messages[2].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Third page:
|
|
require.False(t, response.IsComplete())
|
|
require.Len(t, response.Messages(), 1)
|
|
require.Equal(t, response.Messages()[0].Message.GetTimestamp(), messages[0].GetTimestamp())
|
|
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
|
|
// -- Trying to continue a completed cursor
|
|
err = response.Next(ctx)
|
|
require.NoError(t, err)
|
|
require.True(t, response.IsComplete())
|
|
|
|
// No cursor should be returned if there are no messages that match the criteria
|
|
response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "no-messages"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(true, 2))
|
|
require.NoError(t, err)
|
|
require.Len(t, response.Messages(), 0)
|
|
require.Empty(t, response.Cursor())
|
|
|
|
// If the page size is larger than the number of existing messages, it should not return a cursor
|
|
response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithPaging(true, 100))
|
|
require.NoError(t, err)
|
|
require.Len(t, response.Messages(), 5)
|
|
require.Empty(t, response.Cursor())
|
|
|
|
// Invalid cursors should fail
|
|
// TODO: nwaku does not support this feature yet
|
|
//_, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithCursor([]byte{1, 2, 3, 4, 5, 6}))
|
|
//require.Error(t, err)
|
|
|
|
// Inexistent cursors should return an empty response
|
|
// TODO: nwaku does not support this feature yet
|
|
//response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: endTime}, WithCursor(make([]byte, 32))) // Requesting cursor 0x00...00
|
|
//require.NoError(t, err)
|
|
//require.Len(t, response.messages, 0)
|
|
//require.Empty(t, response.Cursor())
|
|
|
|
// Handle temporal history query with an invalid time window
|
|
_, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: endTime, TimeEnd: startTime})
|
|
require.NotNil(t, err)
|
|
|
|
// Handle temporal history query with a zero-size time window
|
|
response, err = wakuStore.Query(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test"), TimeStart: startTime, TimeEnd: startTime})
|
|
require.NoError(t, err)
|
|
require.Len(t, response.Messages(), 0)
|
|
require.Empty(t, response.Cursor())
|
|
|
|
// Should not include data
|
|
response, err = wakuStore.Request(ctx, MessageHashCriteria{MessageHashes: []pb.MessageHash{messages[0].Hash(pubsubTopic)}}, IncludeData(false), WithPeer(storenode.ID))
|
|
require.NoError(t, err)
|
|
require.Len(t, response.Messages(), 1)
|
|
require.Nil(t, response.Messages()[0].Message)
|
|
|
|
response, err = wakuStore.Request(ctx, FilterCriteria{ContentFilter: protocol.NewContentFilter(pubsubTopic, "test")}, IncludeData(false))
|
|
require.NoError(t, err)
|
|
require.GreaterOrEqual(t, len(response.Messages()), 1)
|
|
require.Nil(t, response.Messages()[0].Message)
|
|
}
|