Merge pull request #118 from Warp-net/develop

Develop

Author: filinvadim

.githooks/pre-commit

@@ -34,6 +34,9 @@ NEW_VERSION="$MAJOR.$MINOR.$PATCH"
 sed -i "s/^$CURRENT_VERSION\$/$NEW_VERSION/" "$CONFIG"
 sed -i -E "s/^version:[[:space:]]*'?([0-9]+\.[0-9]+\.[0-9]+)'?/version: ${NEW_VERSION}/" "$SNAPCRAFT"
 
+go mod tidy
+go mod vendor
+
 git add "$CONFIG" "$SNAPCRAFT"
 
 

.github/workflows/check-static.yaml

@@ -5,12 +5,12 @@ env:
 
 on:
   workflow_dispatch:
-#  push:
-#    branches:
-#      - main
-#  pull_request:
-#    branches:
-#      - main
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
 
 jobs:
   lint:

cmd/node/bootstrap/socks5/server.go

@@ -33,7 +33,9 @@ type MetricsPusher interface {
 	RemoveSocksConnections(ip string)
 }
 
-const reqIpKey = "ip-key"
+type ctxKey string
+
+const reqIpKey ctxKey = "ip-key"
 
 type rule struct {
 	m MetricsPusher
@@ -133,13 +135,13 @@ func (s *socksServer) warpnetOverlayHandler(ctx context.Context, proto, addr str
 
 	peer, isRedirect := s.balancer.route()
 	if peer == "" {
-		return nil, fmt.Errorf("no peers found") //nolint:errcheck
+		return nil, fmt.Errorf("no peers found") //nolint:err113
 	}
 	if isRedirect {
 		peerAddrs := s.streamer.Peerstore().Addrs(peer)
 		log.Infof("socks5 server: redirect to %v", peerAddrs)
 		for _, pAddr := range peerAddrs {
-			conn, err := net.DialTimeout(proto, toNetAddr(pAddr).String(), time.Second)
+			conn, err := net.DialTimeout(proto, toNetAddr(pAddr).String(), time.Second) //nolint:noctx
 			if err != nil {
 				continue
 			}

cmd/node/member/node/member-node.go

@@ -64,7 +64,7 @@ type MemberNode struct {
 	pubsubService        PubSubProvider
 	dHashTable           DistributedHashTableCloser
 	nodeRepo             NodeProvider
-	statsRepo            NodeProvider
+	statsRepo            StatsProvider
 	authRepo             AuthProvider
 	userRepo             UserProvider
 	followRepo           FollowStorer
@@ -88,13 +88,13 @@ func NewMemberNode(
 	if len(privKey) == 0 {
 		return nil, node.ErrPrivateKeyRequired
 	}
-	nodeRepo := database.NewNodeRepo(db, "NODES")
+	nodeRepo := database.NewNodeRepo(db)
 	store, err := warpnet.NewPeerstore(ctx, nodeRepo)
 	if err != nil {
 		return nil, err
 	}
 
-	statsRepo := database.NewNodeRepo(db, "CRDT")
+	statsRepo := database.NewStatsRepo(db)
 	userRepo := database.NewUserRepo(db)
 	followRepo := database.NewFollowRepo(db)
 	owner := authRepo.GetOwner()

cmd/node/member/node/types.go

@@ -75,7 +75,10 @@ type DistributedHashTableCloser interface {
 
 type NodeProvider interface {
 	datastore.Datastore
-	Prefix() string
+}
+
+type StatsProvider interface {
+	datastore.Datastore
 }
 
 type AuthProvider interface {

core/crdt/stats.go

@@ -58,7 +58,6 @@ type Broadcaster interface {
 
 type CRDTStorer interface {
 	ds.Datastore
-	Prefix() string
 }
 
 type CRDTRouter interface {
@@ -84,10 +83,6 @@ func NewCRDTStatsStore(
 	node host.Host,
 	router CRDTRouter,
 ) (*CRDTStatsStore, error) {
-	prefix := datastore.Prefix()
-	if prefix != "/CRDT" {
-		return nil, warpnet.WarpError("CRDT datastore namespace must start with '/CRDT' prefix")
-	}
 	ctx, cancel := context.WithCancel(ctx)
 
 	baseStore := ds.MutexWrap(datastore)

core/transport/integration_test.go

@@ -641,7 +641,6 @@ func TestIntegration_MDNSDiscovery(t *testing.T) {
 		libp2p.DisableRelay(),
 	)
 	require.NoError(t, err)
-	defer hostA.Close()
 
 	// Host B.
 	hostB, err := libp2p.New(
@@ -653,7 +652,6 @@ func TestIntegration_MDNSDiscovery(t *testing.T) {
 		libp2p.DisableRelay(),
 	)
 	require.NoError(t, err)
-	defer hostB.Close()
 
 	t.Logf("Host A: %s addrs=%v", hostA.ID(), hostA.Addrs())
 	t.Logf("Host B: %s addrs=%v", hostB.ID(), hostB.Addrs())
@@ -681,7 +679,6 @@ func TestIntegration_MDNSDiscovery(t *testing.T) {
 	}
 	mdnsA := mdns.NewMdnsService(hostA, mdnsServiceName, notifeeA)
 	require.NoError(t, mdnsA.Start())
-	defer mdnsA.Close()
 
 	notifeeB := &mdnsNotifee{
 		h:         hostB,
@@ -690,7 +687,15 @@ func TestIntegration_MDNSDiscovery(t *testing.T) {
 	}
 	mdnsB := mdns.NewMdnsService(hostB, mdnsServiceName, notifeeB)
 	require.NoError(t, mdnsB.Start())
-	defer mdnsB.Close()
+	defer func() {
+		mdnsA.Close()
+		mdnsB.Close()
+
+		time.Sleep(100 * time.Millisecond)
+
+		hostA.Close()
+		hostB.Close()
+	}()
 
 	// Wait for at least one side to discover and connect to the other.
 	t.Log("Waiting for mDNS discovery...")
@@ -765,11 +770,9 @@ func TestIntegration_MDNSDiscoveryWithPSKAndRelay(t *testing.T) {
 
 	hostA, err := libp2p.New(commonOpts()...)
 	require.NoError(t, err)
-	defer hostA.Close()
 
 	hostB, err := libp2p.New(commonOpts()...)
 	require.NoError(t, err)
-	defer hostB.Close()
 
 	t.Logf("Host A: %s addrs=%v", hostA.ID(), hostA.Addrs())
 	t.Logf("Host B: %s addrs=%v", hostB.ID(), hostB.Addrs())
@@ -797,7 +800,6 @@ func TestIntegration_MDNSDiscoveryWithPSKAndRelay(t *testing.T) {
 	}
 	mdnsA := mdns.NewMdnsService(hostA, mdnsServiceName, notifeeA)
 	require.NoError(t, mdnsA.Start())
-	defer mdnsA.Close()
 
 	notifeeB := &mdnsNotifee{
 		h:         hostB,
@@ -806,8 +808,16 @@ func TestIntegration_MDNSDiscoveryWithPSKAndRelay(t *testing.T) {
 	}
 	mdnsB := mdns.NewMdnsService(hostB, mdnsServiceName, notifeeB)
 	require.NoError(t, mdnsB.Start())
-	defer mdnsB.Close()
 
+	defer func() {
+		mdnsA.Close()
+		mdnsB.Close()
+
+		time.Sleep(100 * time.Millisecond)
+
+		hostA.Close()
+		hostB.Close()
+	}()
 	t.Log("Waiting for mDNS discovery with PSK+Relay...")
 	select {
 	case pid := <-notifeeA.connected:

database/node-repo.go

@@ -44,6 +44,7 @@ import (
 // slash is required because of: invalid datastore key: NODES:/peers/keys/AASAQAISEAXNRKHMX2O3AA26JM7NGIWUPOGIITJ2UHHXGX4OWIEKPNAW6YCSK/priv
 const (
 	requiredPrefixSlash = "/"
+	nodesPrefix         = "NODES"
 
 	ErrNilNodeRepo = local_store.DBError("node repo is nil")
 )
@@ -71,7 +72,8 @@ type NodeRepo struct {
 	BootstrapSelfHashHex string
 }
 
-func NewNodeRepo(db NodeStorer, prefix string) *NodeRepo {
+func NewNodeRepo(db NodeStorer) *NodeRepo {
+	prefix := nodesPrefix
 	if !strings.HasPrefix(prefix, requiredPrefixSlash) {
 		prefix = requiredPrefixSlash + prefix
 	}
@@ -83,10 +85,6 @@ func NewNodeRepo(db NodeStorer, prefix string) *NodeRepo {
 	return nr
 }
 
-func (d *NodeRepo) Prefix() string {
-	return d.prefix
-}
-
 func (d *NodeRepo) Put(ctx context.Context, key datastore.Key, value []byte) error {
 	if d == nil || d.db == nil {
 		return ErrNilNodeRepo
@@ -345,9 +343,7 @@ func (d *NodeRepo) query(tx *local_store.Txn, q datastore.Query) (_ datastore.Re
 	opt := local_store.DefaultIteratorOptions
 	opt.PrefetchValues = !q.KeysOnly
 
-	key := strings.TrimPrefix(q.Prefix, "/")
-	prefix := local_store.NewPrefixBuilder(d.prefix).AddRootID(key).Build().Bytes()
-	opt.Prefix = prefix
+	opt.Prefix = d.storageQueryPrefix(q.Prefix)
 
 	// Handle ordering
 	if len(q.Orders) > 0 {
@@ -405,7 +401,7 @@ func (d *NodeRepo) query(tx *local_store.Txn, q datastore.Query) (_ datastore.Re
 			matches := true
 			check := func(value []byte) error {
 				e := datastore.DsEntry{
-					Key:   string(item.Key()),
+					Key:   d.resultKeyFromStorageKey(string(item.Key())),
 					Value: value,
 					Size:  int(item.ValueSize()),
 				}
@@ -443,7 +439,7 @@ func (d *NodeRepo) query(tx *local_store.Txn, q datastore.Query) (_ datastore.Re
 				return
 			}
 			item := it.Item()
-			e := datastore.DsEntry{Key: string(item.Key())}
+			e := datastore.DsEntry{Key: d.resultKeyFromStorageKey(string(item.Key()))}
 
 			var result datastore.Result
 			if !q.KeysOnly {
@@ -481,6 +477,30 @@ func (d *NodeRepo) query(tx *local_store.Txn, q datastore.Query) (_ datastore.Re
 	return results, nil
 }
 
+func (d *NodeRepo) storageQueryPrefix(queryPrefix string) []byte {
+	prefix := strings.TrimSuffix(datastore.NewKey(queryPrefix).String(), requiredPrefixSlash)
+	base := strings.TrimSuffix(d.prefix, requiredPrefixSlash)
+
+	if prefix == "" {
+		return []byte(base + requiredPrefixSlash)
+	}
+
+	return []byte(base + prefix + requiredPrefixSlash)
+}
+
+func (d *NodeRepo) resultKeyFromStorageKey(storageKey string) string {
+	if storageKey == d.prefix {
+		return requiredPrefixSlash
+	}
+
+	trimPrefix := d.prefix + requiredPrefixSlash
+	if strings.HasPrefix(storageKey, trimPrefix) { //nolint:modernize
+		return requiredPrefixSlash + strings.TrimPrefix(storageKey, trimPrefix)
+	}
+
+	return storageKey
+}
+
 func filter(filters []datastore.Filter, entry datastore.DsEntry) bool {
 	for _, f := range filters {
 		if !f.Filter(entry) {

database/node-repo_test.go

@@ -59,7 +59,7 @@ func (s *NodeRepoTestSuite) SetupSuite() {
 	auth := NewAuthRepo(s.db)
 	s.Require().NoError(auth.Authenticate("test", "test"))
 
-	s.repo = NewNodeRepo(s.db, "test")
+	s.repo = NewNodeRepo(s.db)
 
 }
 
@@ -112,12 +112,12 @@ func (s *NodeRepoTestSuite) TestDiskUsage() {
 }
 
 func (s *NodeRepoTestSuite) TestQuerySimple() {
-	key := datastore.NewKey("query/key")
+	key := datastore.NewKey("querysimple/key/item")
 	val := []byte("qval")
 	err := s.repo.Put(s.ctx, key, val)
 	s.Require().NoError(err)
 
-	q := query.Query{Prefix: "query/key"}
+	q := query.Query{Prefix: "querysimple/key"}
 	results, err := s.repo.Query(s.ctx, q)
 	s.Require().NoError(err)
 	s.Require().NotNil(results)
@@ -130,12 +130,64 @@ func (s *NodeRepoTestSuite) TestQuerySimple() {
 		if r.Error != nil {
 			continue
 		}
+		s.Equal("/querysimple/key/item", r.Entry.Key)
 		found = true
 		break
 	}
 	s.True(found)
 }
 
+func (s *NodeRepoTestSuite) TestQueryEmptyPrefix() {
+	key := datastore.NewKey("all/key")
+	val := []byte("all")
+	err := s.repo.Put(s.ctx, key, val)
+	s.Require().NoError(err)
+
+	results, err := s.repo.Query(s.ctx, query.Query{})
+	s.Require().NoError(err)
+	s.Require().NotNil(results)
+
+	defer func() {
+		_ = results.Close()
+	}()
+
+	var found bool
+	for r := range results.Next() {
+		if r.Error != nil {
+			continue
+		}
+		if r.Entry.Key == "/all/key" {
+			found = true
+			break
+		}
+	}
+
+	s.True(found)
+}
+
+func (s *NodeRepoTestSuite) TestQueryPrefixDoesNotMatchSiblingKeys() {
+	err := s.repo.Put(s.ctx, datastore.NewKey("query/key/child"), []byte("child"))
+	s.Require().NoError(err)
+	err = s.repo.Put(s.ctx, datastore.NewKey("query/key2"), []byte("sibling"))
+	s.Require().NoError(err)
+
+	results, err := s.repo.Query(s.ctx, query.Query{Prefix: "query/key"})
+	s.Require().NoError(err)
+	s.Require().NotNil(results)
+	defer func() {
+		_ = results.Close()
+	}()
+
+	keys := make([]string, 0)
+	for r := range results.Next() {
+		s.Require().NoError(r.Error)
+		keys = append(keys, r.Entry.Key)
+	}
+
+	s.Contains(keys, "/query/key/child")
+	s.NotContains(keys, "/query/key2")
+}
+
 func TestNodeRepoTestSuite(t *testing.T) {
 	defer goleak.VerifyNone(t)