lighthouse: Never block in QueryServer, drop queries when the channel is full (#1910)

connection_manager_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/rcrowley/go-metrics"
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/config"
 	"github.com/slackhq/nebula/overlay/overlaytest"
@@ -21,6 +22,8 @@ func newTestLighthouse() *LightHouse {
 		l:         test.NewLogger(),
 		addrMap:   map[netip.Addr]*RemoteList{},
 		queryChan: make(chan netip.Addr, 10),
+		// Nothing drains queryChan here, so an 11th query is dropped.
+		queryDropped: metrics.NilCounter{},
 	}
 	lighthouses := []netip.Addr{}
 	staticList := map[netip.Addr]struct{}{}

examples/config.yml

@@ -410,8 +410,8 @@ logging:
   #try_interval: 100ms
   #retries: 20
 
-  # query_buffer is the size of the buffer channel for querying lighthouses
-  #query_buffer: 64
+  # query_buffer is the size of the buffer channel for querying lighthouses, drops are recorded in the lighthouse.query.dropped metric.
+  #query_buffer: 1024
 
   # trigger_buffer is the size of the buffer channel for quickly sending handshakes
   # after receiving the response for lighthouse queries

lighthouse.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/gaissmai/bart"
+	"github.com/rcrowley/go-metrics"
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/config"
 	"github.com/slackhq/nebula/header"
@@ -81,6 +82,8 @@ type LightHouse struct {
 
 	updateTrigger chan struct{}
 	queryChan     chan netip.Addr
+	// queryDropped counts queries QueryServer dropped because queryChan was full.
+	queryDropped metrics.Counter
 
 	calculatedRemotes atomic.Pointer[bart.Table[[]*calculatedRemote]] // Maps VpnAddr to []*calculatedRemote
 
@@ -116,7 +119,8 @@ func NewLightHouseFromConfig(ctx context.Context, l *slog.Logger, c *config.C, c
 		nebulaPort:         nebulaPort,
 		punchy:             p,
 		updateTrigger:      make(chan struct{}, 1),
-		queryChan:          make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 64)),
+		queryChan:          make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 1024)),
+		queryDropped:       metrics.GetOrRegisterCounter("lighthouse.query.dropped", nil),
 		l:                  l,
 	}
 	h.localAddrsFn = func(al *LocalAllowList) []netip.Addr {
@@ -512,14 +516,19 @@ func (lh *LightHouse) Query(vpnAddr netip.Addr) *RemoteList {
 	return nil
 }
 
-// QueryServer is asynchronous so no reply should be expected
+// QueryServer is asynchronous and non-blocking, no reply should be expected.
+// If the query channel is full, the query is dropped.
 func (lh *LightHouse) QueryServer(vpnAddr netip.Addr) {
 	// Don't put lighthouse addrs in the query channel because we can't query lighthouses about lighthouses
 	if lh.amLighthouse || lh.IsLighthouseAddr(vpnAddr) {
 		return
 	}
 
-	lh.queryChan <- vpnAddr
+	select {
+	case lh.queryChan <- vpnAddr:
+	default:
+		lh.queryDropped.Inc(1)
+	}
 }
 
 func (lh *LightHouse) QueryCache(vpnAddrs []netip.Addr) *RemoteList {

lighthouse_test.go

@@ -6,7 +6,9 @@ import (
 	"errors"
 	"fmt"
 	"net/netip"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/cert"
@@ -795,3 +797,55 @@ func TestLightHouse_logLocalAddrsErr(t *testing.T) {
 	lh.logLocalAddrsErr(errors.New("something else"))
 	assert.Contains(t, out.String(), "level=WARN")
 }
+
+// heldEncWriter holds the query worker in its first GetHostInfo until release closes.
+type heldEncWriter struct {
+	mockEncWriter
+	once    sync.Once
+	held    chan struct{}
+	release chan struct{}
+}
+
+func (w *heldEncWriter) GetHostInfo(netip.Addr) *HostInfo {
+	w.once.Do(func() { close(w.held) })
+	<-w.release
+	return nil
+}
+
+// QueryServer never waits on the query worker, since the packet readers call it and nothing drains the channel at
+// shutdown; a query that finds the channel full is dropped and counted. The worker is held on its first query so the
+// channel fills.
+func TestLighthouse_QueryServerDoesNotBlock(t *testing.T) {
+	l := test.NewLogger()
+	cs := testCertState(netip.MustParsePrefix("10.128.0.1/16"))
+	lh1 := "10.128.0.2"
+	c := config.NewC(l)
+	c.Settings["lighthouse"] = map[string]any{"hosts": []any{lh1}}
+	c.Settings["static_host_map"] = map[string]any{lh1: []any{"1.1.1.1:4242"}}
+	c.Settings["handshakes"] = map[string]any{"query_buffer": 1}
+	lh, err := NewLightHouseFromConfig(t.Context(), l, c, cs, nil, nil)
+	require.NoError(t, err)
+	w := &heldEncWriter{held: make(chan struct{}), release: make(chan struct{})}
+	lh.ifce = w
+	t.Cleanup(func() { close(w.release) })
+
+	lh.QueryServer(netip.MustParseAddr("10.128.0.10"))
+	select {
+	case <-w.held:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the query worker never took the first query")
+	}
+	dropped := lh.queryDropped.Count()
+	done := make(chan struct{})
+	go func() {
+		lh.QueryServer(netip.MustParseAddr("10.128.0.11"))
+		lh.QueryServer(netip.MustParseAddr("10.128.0.12"))
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("QueryServer blocked on a full query channel")
+	}
+	assert.EqualValues(t, 1, lh.queryDropped.Count()-dropped)
+}