feat(dir): dht only remote discovery

cli/cmd/daemon/daemon.config.yaml

@@ -57,17 +57,6 @@ server:
     # more announcement traffic and more work for every subscriber.
     republish_interval: "36h"
 
-    gossipsub:
-      enabled: true
-    # DHT-based record + referrer autosync (deny-by-default; disabled unless set).
-    # When enabled, records announced by an allow-listed peer are pulled and
-    # ingested locally over the libp2p/DHT transport.
-    autosync:
-      enabled: false
-      # peerlist is a list of trusted source peers (by libp2p peer ID).
-      # peerlist:
-      #   - peer: "12D3KooW...peerID1"
-      #   - peer: "12D3KooW...peerID2"
     # Circuit-relay v2 for NAT traversal.
     # relay_service: enable a relay service on this node (only on publicly
     #   reachable nodes, e.g. bootstrap) so it can relay traffic for NAT'd peers.

cli/cmd/doctor/bootstrap_checks.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	serverrouting "github.com/agntcy/dir/server/routing"
 	"github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -246,9 +247,8 @@ func addPeerProtocolDetails(details map[string]string, protocols []protocol.ID,
 
 	details["protocol_count"] = fmt.Sprintf("%d", len(protocolStrings))
 	details["protocols"] = strings.Join(protocolStrings, ",")
-	details["has_kad_dht_protocol"] = fmt.Sprintf("%t", hasProtocolPrefix(protocolStrings, "/ipfs/kad") || hasProtocolPrefix(protocolStrings, "dir/kad"))
+	details["has_kad_dht_protocol"] = fmt.Sprintf("%t", hasProtocolPrefix(protocolStrings, "/ipfs/kad") || hasProtocolPrefix(protocolStrings, serverrouting.ProtocolPrefix+"/kad"))
 	details["has_dir_rpc_protocol"] = fmt.Sprintf("%t", hasProtocolPrefix(protocolStrings, "/dir/rpc"))
-	details["has_gossipsub_protocol"] = fmt.Sprintf("%t", hasProtocolPrefix(protocolStrings, "/meshsub") || hasProtocolPrefix(protocolStrings, "/floodsub"))
 }
 
 func hasProtocolPrefix(protocols []string, prefix string) bool {

cli/cmd/doctor/bootstrap_checks_test.go

@@ -108,13 +108,12 @@ func TestBootstrapPeerValidationHelpers(t *testing.T) {
 func TestAddPeerProtocolDetails(t *testing.T) {
 	details := map[string]string{}
 
-	addPeerProtocolDetails(details, []protocol.ID{"/ipfs/kad/1.0.0", "/dir/rpc/0.1.0", "/meshsub/1.1.0"}, nil)
+	addPeerProtocolDetails(details, []protocol.ID{"/dir/2/kad/1.0.0", "/dir/rpc/2.0.0"}, nil)
 
-	assert.Equal(t, "3", details["protocol_count"])
+	assert.Equal(t, "2", details["protocol_count"])
 	assert.Equal(t, "true", details["has_kad_dht_protocol"])
 	assert.Equal(t, "true", details["has_dir_rpc_protocol"])
-	assert.Equal(t, "true", details["has_gossipsub_protocol"])
-	assert.True(t, hasProtocolPrefix([]string{"/dir/rpc/0.1.0"}, "/dir/rpc"))
+	assert.True(t, hasProtocolPrefix([]string{"/dir/rpc/2.0.0"}, "/dir/rpc"))
 	assert.False(t, hasProtocolPrefix([]string{"/other/1.0.0"}, "/dir/rpc"))
 }
 

cli/go.mod

@@ -520,7 +520,6 @@ require (
 	github.com/libp2p/go-libp2p-gorpc v0.6.0 // indirect
 	github.com/libp2p/go-libp2p-kad-dht v0.41.0
 	github.com/libp2p/go-libp2p-kbucket v0.8.0 // indirect
-	github.com/libp2p/go-libp2p-pubsub v0.16.0 // indirect
 	github.com/libp2p/go-libp2p-record v0.3.1 // indirect
 	github.com/libp2p/go-libp2p-routing-helpers v0.7.5 // indirect
 	github.com/libp2p/go-msgio v0.3.0 // indirect

cli/go.sum

@@ -1162,8 +1162,6 @@ github.com/libp2p/go-libp2p-kad-dht v0.41.0 h1:sDigz5SgV20Crj8ItJmJpEAM+eJrzC/Sa
 github.com/libp2p/go-libp2p-kad-dht v0.41.0/go.mod h1:2qc4QGLvmIdznYbNg++FF76vp4q2SaBZyr76jHV8xgs=
 github.com/libp2p/go-libp2p-kbucket v0.8.0 h1:QAK7RzKJpYe+EuSEATAaaHYMYLkPDGC18m9jxPLnU8s=
 github.com/libp2p/go-libp2p-kbucket v0.8.0/go.mod h1:JMlxqcEyKwO6ox716eyC0hmiduSWZZl6JY93mGaaqc4=
-github.com/libp2p/go-libp2p-pubsub v0.16.0 h1:j7G2C8kJwkcAQqYR7Wmq3d75d3Sgw/N0Hhiv0dVx7OY=
-github.com/libp2p/go-libp2p-pubsub v0.16.0/go.mod h1:lr4oE8bFgQaifRcoc2uWhWWiK6tPdOEKpUuR408GFN4=
 github.com/libp2p/go-libp2p-record v0.3.1 h1:cly48Xi5GjNw5Wq+7gmjfBiG9HCzQVkiZOUZ8kUl+Fg=
 github.com/libp2p/go-libp2p-record v0.3.1/go.mod h1:T8itUkLcWQLCYMqtX7Th6r7SexyUJpIyPgks757td/E=
 github.com/libp2p/go-libp2p-routing-helpers v0.7.5 h1:HdwZj9NKovMx0vqq6YNPTh6aaNzey5zHD7HeLJtq6fI=

docs/content/dir/dir-cli-reference.md

@@ -312,8 +312,6 @@ The daemon ships with sensible built-in defaults. To customize, pass a YAML conf
       routing:
         listen_address: "/ip4/0.0.0.0/tcp/8999"
         datastore_dir: "routing"
-        gossipsub:
-          enabled: true
       database:
         type: "sqlite"
         sqlite:

docs/content/dir/dir-deployment-kubernetes.md

@@ -124,8 +124,6 @@ The Agent Directory Service can be deployed using Helm or GitOps / Argo CD. Helm
               listen_address: "/ip4/0.0.0.0/tcp/5555"
               datastore_dir: /etc/routing/datastore
               directory_api_address: "dir-apiserver.dir-dev-dir.svc.cluster.local:8888"
-              gossipsub:
-                enabled: false
             sync:
               auth_config:
                 username: "user"

docs/content/dir/dir-deployment-local.md

@@ -144,8 +144,6 @@ server:
     datastore_dir: "routing"
     bootstrap_peers:
       - "/dns4/remote-dir.example.com/tcp/8999/p2p/<remote-peer-id>"
-    gossipsub:
-      enabled: true
   database:
     type: "sqlite"
     sqlite:

docs/content/dir/dir-federation-aws-eks.md

@@ -382,8 +382,6 @@ This guide does not try to provision the AWS infrastructure from zero in the mai
           key_path: /etc/routing/node.privkey
           datastore_dir: /etc/routing/datastore
           directory_api_address: "${DIR_API_HOST}:443"
-          gossipsub:
-            enabled: true
         sync:
           auth_config:
             username: "user"

install/charts/dir/apiserver/values.yaml

@@ -99,13 +99,6 @@ config:
     bootstrap_peers:
       - /dns4/routing.ads.outshift.io/tcp/5555/p2p/12D3KooWLf9p3cedc86xGQBaqak6rAFmQk1HxKAK1yh7umHE3amu
 
-    # GossipSub configuration for efficient label announcements
-    # When enabled, labels are propagated via GossipSub mesh to ALL subscribed peers
-    # When disabled, falls back to DHT+Pull mechanism (higher bandwidth, limited reach)
-    # Default: true (recommended for production)
-    gossipsub:
-      enabled: true
-
   # Sync configuration
   sync:
     # Authentication configuration for sync operations

install/charts/dir/values.yaml

@@ -203,24 +203,6 @@ apiserver:
       bootstrap_peers:
         - /dns4/routing.ads.outshift.io/tcp/5555/p2p/12D3KooWLf9p3cedc86xGQBaqak6rAFmQk1HxKAK1yh7umHE3amu
 
-      # GossipSub configuration for efficient label announcements
-      # When enabled, labels are propagated via GossipSub mesh to ALL subscribed peers
-      # When disabled, falls back to DHT+Pull mechanism (higher bandwidth, limited reach)
-      # Default: true (recommended for production)
-      gossipsub:
-        enabled: true
-
-      # DHT-based record + referrer autosync.
-      # Deny-by-default: when enabled, only records announced by a peer in
-      # peerlist are pulled and ingested locally over the libp2p/DHT transport.
-      # peerlist is a list of trusted source peers (by libp2p peer ID) and is
-      # configured via values/config only (not a single environment variable).
-      autosync:
-        enabled: false
-        # peerlist:
-        #   - peer: "12D3KooW...peerID1"
-        #   - peer: "12D3KooW...peerID2"
-
       # Circuit-relay v2 for NAT traversal.
       # relay_service: run a relay service on this node so it can relay traffic
       #   for NAT'd peers. Enable only on publicly-reachable nodes (e.g. the

install/docker/apiserver.env

@@ -40,7 +40,6 @@ DIRECTORY_SERVER_ROUTING_DIRECTORY_API_ADDRESS=
 DIRECTORY_SERVER_ROUTING_BOOTSTRAP_PEERS=
 DIRECTORY_SERVER_ROUTING_KEY_PATH=
 DIRECTORY_SERVER_ROUTING_DATASTORE_DIR=
-DIRECTORY_SERVER_ROUTING_GOSSIPSUB_ENABLED=true
 
 # Database Configuration (PostgreSQL)
 DIRECTORY_SERVER_DATABASE_TYPE=postgres

reconciler/tasks/metrics/task.go

@@ -12,6 +12,7 @@ package metrics
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/agntcy/dir/server/types"
@@ -20,6 +21,26 @@ import (
 
 var logger = logging.Logger("reconciler/metrics")
 
+// providerCountWorkers bounds how many provider lookups are in flight at once.
+const providerCountWorkers = 8
+
+// maxProviderCount is an upper bound for a stored provider count. A real DHT
+// lookup returns orders of magnitude fewer peers; the cap exists so the
+// conversion to the column's width cannot wrap.
+const maxProviderCount = 1 << 20
+
+func clampProviderCount(count int) uint32 {
+	if count < 0 {
+		return 0
+	}
+
+	if count > maxProviderCount {
+		return maxProviderCount
+	}
+
+	return uint32(count)
+}
+
 // ProviderCounterAPI is the minimal interface required by the metrics task to
 // query provider counts. It is satisfied by types.RoutingAPI (daemon mode, where
 // the routing layer is shared in-process) and by GRPCProviderCounter (standalone
@@ -96,28 +117,21 @@ func (t *Task) refreshProviderCounts(ctx context.Context) error {
 
 	var updated, failed int
 
-	for _, cid := range cids {
-		select {
-		case <-ctx.Done():
-			return fmt.Errorf("context cancelled: %w", ctx.Err())
-		default:
-		}
-
-		count, err := t.counters.GetProviderCount(ctx, cid)
-		if err != nil {
-			logger.Warn("Failed to get provider count", "cid", cid, "error", err)
+	// Persist serially while the lookups run concurrently. The lookups are the
+	// slow part; the metrics table serialises writers regardless.
+	for res := range t.countProviders(ctx, cids) {
+		if res.err != nil {
+			logger.Warn("Failed to get provider count", "cid", res.cid, "error", res.err)
 
 			failed++
 
 			continue
 		}
 
-		if count < 0 {
-			count = 0
-		}
+		count := clampProviderCount(res.count)
 
-		if err := t.db.SetProviderCount(cid, uint32(count)); err != nil {
-			logger.Warn("Failed to set provider count", "cid", cid, "count", count, "error", err)
+		if err := t.db.SetProviderCount(res.cid, count); err != nil {
+			logger.Warn("Failed to set provider count", "cid", res.cid, "count", count, "error", err)
 
 			failed++
 
@@ -127,7 +141,62 @@ func (t *Task) refreshProviderCounts(ctx context.Context) error {
 		updated++
 	}
 
+	if err := ctx.Err(); err != nil {
+		return fmt.Errorf("context cancelled: %w", err)
+	}
+
 	logger.Info("Provider count refresh complete", "updated", updated, "failed", failed)
 
 	return nil
 }
+
+type providerCountResult struct {
+	cid   string
+	count int
+	err   error
+}
+
+// countProviders fans the CID list out over a bounded worker pool and streams
+// results back as they land. Each GetProviderCount is a DHT lookup, so running
+// one CID at a time would exceed the reconciliation interval on any sizeable
+// corpus.
+func (t *Task) countProviders(ctx context.Context, cids []string) <-chan providerCountResult {
+	jobs := make(chan string)
+	results := make(chan providerCountResult)
+	workers := min(providerCountWorkers, len(cids))
+
+	var wg sync.WaitGroup
+
+	for range workers {
+		wg.Go(func() {
+			for cid := range jobs {
+				count, err := t.counters.GetProviderCount(ctx, cid)
+
+				select {
+				case results <- providerCountResult{cid: cid, count: count, err: err}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		})
+	}
+
+	go func() {
+		defer close(jobs)
+
+		for _, cid := range cids {
+			select {
+			case jobs <- cid:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	go func() {
+		wg.Wait()
+		close(results)
+	}()
+
+	return results
+}