client/daemon: route liveness probing

Author: snormore

.golangci.yaml

@@ -74,6 +74,9 @@ linters:
       - third_party$
       - builtin$
       - examples$
+    rules:
+      - linters: [govet]
+        text: 'buildtag: \+build line is no longer needed'
 formatters:
   enable:
     - gofmt

CHANGELOG.md

@@ -8,7 +8,10 @@ All notable changes to this project will be documented in this file.
 
 ### Changes
 
--- Add automated compatibility tests in CI to validate all actual testnet and mainnet state against the current codebase, ensuring backward compatibility across protocol versions.
+- Onchain programs
+    - Add automated compatibility tests in CI to validate all actual testnet and mainnet state against the current codebase, ensuring backward compatibility across protocol versions.
+- Client
+    - Add initial opt-in route liveness probing support to `doublezerod` via `--route-probing-enable` flag (not yet considered stable)
 
 ## [v0.6.9](https://github.com/malbeclabs/doublezero/compare/client/v0.6.7...client/v0.6.9) – 2025-10-24
 

client/doublezerod/cmd/doublezerod/main.go

@@ -1,3 +1,5 @@
+//go:build linux
+
 package main
 
 import (
@@ -11,27 +13,49 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/api"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/bgp"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/manager"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/pim"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/probing"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/routing"
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/runtime"
+	"github.com/malbeclabs/doublezero/client/doublezerod/internal/services"
 	"github.com/malbeclabs/doublezero/config"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
+const (
+	defaultRouteProbingInterval       = 1 * time.Second
+	defaultRouteProbingMaxConcurrency = 1024
+	defaultRouteProbingProbeTimeout   = 1 * time.Second
+	defaultRouteProbingUpThreshold    = 3
+	defaultRouteProbingDownThreshold  = 3
+)
+
 var (
-	sockFile             = flag.String("sock-file", "/var/run/doublezerod/doublezerod.sock", "path to doublezerod domain socket")
-	enableLatencyProbing = flag.Bool("latency-probing", true, "enable latency probing to doublezero nodes")
-	versionFlag          = flag.Bool("version", false, "build version")
-	env                  = flag.String("env", config.EnvTestnet, "environment to use")
-	programId            = flag.String("program-id", "", "override smartcontract program id to monitor")
-	rpcEndpoint          = flag.String("solana-rpc-endpoint", "", "override solana rpc endpoint url")
-	probeInterval        = flag.Int("probe-interval", 30, "latency probe interval in seconds")
-	cacheUpdateInterval  = flag.Int("cache-update-interval", 30, "latency cache update interval in seconds")
-	enableVerboseLogging = flag.Bool("v", false, "enables verbose logging")
-	enableLatencyMetrics = flag.Bool("enable-latency-metrics", false, "enables latency metrics")
-	metricsEnable        = flag.Bool("metrics-enable", false, "Enable prometheus metrics")
-	metricsAddr          = flag.String("metrics-addr", "localhost:0", "Address to listen on for prometheus metrics")
+	sockFile                   = flag.String("sock-file", "/var/run/doublezerod/doublezerod.sock", "path to doublezerod domain socket")
+	enableLatencyProbing       = flag.Bool("latency-probing", true, "enable latency probing to doublezero nodes")
+	versionFlag                = flag.Bool("version", false, "build version")
+	env                        = flag.String("env", config.EnvTestnet, "environment to use")
+	programId                  = flag.String("program-id", "", "override smartcontract program id to monitor")
+	rpcEndpoint                = flag.String("solana-rpc-endpoint", "", "override solana rpc endpoint url")
+	probeInterval              = flag.Int("probe-interval", 30, "latency probe interval in seconds")
+	cacheUpdateInterval        = flag.Int("cache-update-interval", 30, "latency cache update interval in seconds")
+	enableVerboseLogging       = flag.Bool("v", false, "enables verbose logging")
+	enableLatencyMetrics       = flag.Bool("enable-latency-metrics", false, "enables latency metrics")
+	metricsEnable              = flag.Bool("metrics-enable", false, "Enable prometheus metrics")
+	metricsAddr                = flag.String("metrics-addr", "localhost:0", "Address to listen on for prometheus metrics")
+	routeProbingEnable         = flag.Bool("route-probing-enable", false, "enables route liveness probing")
+	routeProbingInterval       = flag.Duration("route-probing-interval", defaultRouteProbingInterval, "route liveness probing interval as a duration (i.e. 5s, 10s, 30s)")
+	routeProbingProbeTimeout   = flag.Duration("route-probing-probe-timeout", defaultRouteProbingProbeTimeout, "route liveness probing probe timeout as a duration (i.e. 1s, 3s, 5s)")
+	routeProbingUpThreshold    = flag.Uint("route-probing-up-threshold", defaultRouteProbingUpThreshold, "route liveness probing up threshold")
+	routeProbingDownThreshold  = flag.Uint("route-probing-down-threshold", defaultRouteProbingDownThreshold, "route liveness probing down threshold")
+	routeProbingMaxConcurrency = flag.Uint("route-probing-max-concurrency", defaultRouteProbingMaxConcurrency, "route liveness probing max concurrency")
 
 	// set by LDFLAGS
 	version = "dev"
@@ -114,7 +138,47 @@ func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 
-	if err := runtime.Run(ctx, *sockFile, *enableLatencyProbing, *enableLatencyMetrics, *programId, *rpcEndpoint, *probeInterval, *cacheUpdateInterval); err != nil {
+	db, err := manager.NewDb()
+	if err != nil {
+		slog.Error("error initializing db", "error", err)
+		os.Exit(1)
+	}
+
+	nlr := routing.Netlink{}
+
+	bgps, err := bgp.NewBgpServer(net.IPv4(1, 1, 1, 1))
+	if err != nil {
+		slog.Error("error creating bgp server", "error", err)
+		os.Exit(1)
+	}
+
+	pim := pim.NewPIMServer()
+
+	nlm := manager.NewNetlinkManager(nlr, bgps, db, func(userType api.UserType) (manager.Provisioner, error) {
+		if userType != api.UserTypeIBRL || !*routeProbingEnable {
+			return manager.CreatePassthroughService(userType, bgps, nlr, db, pim)
+		}
+
+		return services.NewIBRLService(bgps, nlr, db, func(iface string, src net.IP) (bgp.RouteManager, error) {
+			if *routeProbingEnable {
+				return probing.NewRouteManager(&probing.Config{
+					Logger:         logger,
+					Context:        ctx,
+					Netlink:        nlr,
+					Liveness:       probing.NewHysteresisLivenessPolicy(*routeProbingUpThreshold, *routeProbingDownThreshold),
+					ListenFunc:     probing.DefaultListenFunc(logger, iface, src),
+					ProbeFunc:      probing.DefaultProbeFunc(logger, iface, *routeProbingProbeTimeout),
+					Interval:       *routeProbingInterval,
+					ProbeTimeout:   *routeProbingProbeTimeout,
+					MaxConcurrency: *routeProbingMaxConcurrency,
+				})
+			} else {
+				return manager.NewNetlinkerPassthroughRouteManager(nlr), nil
+			}
+		}), nil
+	})
+
+	if err := runtime.Run(ctx, nlm, *sockFile, *enableLatencyProbing, *enableLatencyMetrics, *programId, *rpcEndpoint, *probeInterval, *cacheUpdateInterval); err != nil {
 		slog.Error("runtime error", "error", err)
 		os.Exit(1)
 	}

client/doublezerod/internal/api/requests.go

@@ -1,3 +1,5 @@
+//go:build linux
+
 package api
 
 import (

client/doublezerod/internal/bgp/bgp.go

@@ -1,3 +1,5 @@
+//go:build linux
+
 package bgp
 
 import (
@@ -99,28 +101,27 @@ type PeerConfig struct {
 	RouteTable    int
 	FlushRoutes   bool
 	NoInstall     bool
+	RouteManager  RouteManager
 }
 
 type BgpServer struct {
-	server            *corebgp.Server
-	peerStatusChan    chan SessionEvent
-	peerStatus        map[string]Session
-	peerStatusLock    sync.Mutex
-	routeReaderWriter RouteReaderWriter
+	server         *corebgp.Server
+	peerStatusChan chan SessionEvent
+	peerStatus     map[string]Session
+	peerStatusLock sync.Mutex
 }
 
-func NewBgpServer(routerID net.IP, r RouteReaderWriter) (*BgpServer, error) {
+func NewBgpServer(routerID net.IP) (*BgpServer, error) {
 	corebgp.SetLogger(log.Print)
 	srv, err := corebgp.NewServer(netip.MustParseAddr(routerID.String()))
 	if err != nil {
 		return nil, fmt.Errorf("error creating bgp server: %v", err)
 	}
 	return &BgpServer{
-		server:            srv,
-		peerStatusChan:    make(chan SessionEvent),
-		peerStatus:        make(map[string]Session),
-		peerStatusLock:    sync.Mutex{},
-		routeReaderWriter: r,
+		server:         srv,
+		peerStatusChan: make(chan SessionEvent),
+		peerStatus:     make(map[string]Session),
+		peerStatusLock: sync.Mutex{},
 	}, nil
 }
 
@@ -142,7 +143,7 @@ func (b *BgpServer) AddPeer(p *PeerConfig, advertised []NLRI) error {
 	if p.Port != 0 {
 		peerOpts = append(peerOpts, corebgp.WithPort(p.Port))
 	}
-	plugin := NewBgpPlugin(advertised, p.RouteSrc, p.RouteTable, b.peerStatusChan, p.FlushRoutes, p.NoInstall, b.routeReaderWriter)
+	plugin := NewBgpPlugin(advertised, p.RouteSrc, p.RouteTable, b.peerStatusChan, p.FlushRoutes, p.NoInstall, p.RouteManager)
 	err := b.server.AddPeer(corebgp.PeerConfig{
 		RemoteAddress: netip.MustParseAddr(p.RemoteAddress.String()),
 		LocalAS:       p.LocalAs,

client/doublezerod/internal/bgp/bgp_test.go

@@ -1,3 +1,5 @@
+//go:build linux
+
 package bgp_test
 
 import (
@@ -21,28 +23,36 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-type mockRouteReaderWriter struct {
+type mockRouteManager struct {
 	routesAdded   []*routing.Route
 	routesDeleted []*routing.Route
 	routesFlushed []*routing.Route
 	mu            sync.Mutex
 }
 
-func (m *mockRouteReaderWriter) RouteAdd(route *routing.Route) error {
+func (m *mockRouteManager) PeerOnEstablished() error {
+	return nil
+}
+
+func (m *mockRouteManager) PeerOnClose() error {
+	return nil
+}
+
+func (m *mockRouteManager) RouteAdd(route *routing.Route) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.routesAdded = append(m.routesAdded, route)
 	return nil
 }
 
-func (m *mockRouteReaderWriter) RouteDelete(route *routing.Route) error {
+func (m *mockRouteManager) RouteDelete(route *routing.Route) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.routesDeleted = append(m.routesDeleted, route)
 	return nil
 }
 
-func (m *mockRouteReaderWriter) RouteByProtocol(int) ([]*routing.Route, error) {
+func (m *mockRouteManager) RouteByProtocol(int) ([]*routing.Route, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return []*routing.Route{
@@ -58,19 +68,19 @@ func (m *mockRouteReaderWriter) RouteByProtocol(int) ([]*routing.Route, error) {
 	}, nil
 }
 
-func (m *mockRouteReaderWriter) getRoutesAdded() []*routing.Route {
+func (m *mockRouteManager) getRoutesAdded() []*routing.Route {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return append([]*routing.Route(nil), m.routesAdded...)
 }
 
-func (m *mockRouteReaderWriter) getRoutesDeleted() []*routing.Route {
+func (m *mockRouteManager) getRoutesDeleted() []*routing.Route {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return append([]*routing.Route(nil), m.routesDeleted...)
 }
 
-func (m *mockRouteReaderWriter) getRoutesFlushed() []*routing.Route {
+func (m *mockRouteManager) getRoutesFlushed() []*routing.Route {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return append([]*routing.Route(nil), m.routesFlushed...)
@@ -113,8 +123,8 @@ func (p *dummyPlugin) handleUpdate(peer corebgp.PeerConfig, u []byte) *corebgp.N
 }
 
 func TestBgpServer(t *testing.T) {
-	nlr := &mockRouteReaderWriter{}
-	b, err := bgp.NewBgpServer(net.IP{1, 1, 1, 1}, nlr)
+	routeManager := &mockRouteManager{}
+	b, err := bgp.NewBgpServer(net.IP{1, 1, 1, 1})
 	if err != nil {
 		t.Fatalf("error creating bgp server: %v", err)
 	}
@@ -135,6 +145,7 @@ func TestBgpServer(t *testing.T) {
 			FlushRoutes:   true,
 			RouteTable:    syscall.RT_TABLE_MAIN,
 			RouteSrc:      net.IP{7, 7, 7, 7},
+			RouteManager:  routeManager,
 		},
 		[]bgp.NLRI{
 			{AsPath: []uint32{}, NextHop: "1.1.1.1", Prefix: "10.0.0.0", PrefixLength: 32},
@@ -243,7 +254,7 @@ func TestBgpServer(t *testing.T) {
 				Table:   syscall.RT_TABLE_MAIN,
 			},
 		}
-		if diff := checkRoutes(nlr.getRoutesDeleted(), want); diff != "" {
+		if diff := checkRoutes(routeManager.getRoutesDeleted(), want); diff != "" {
 			t.Fatalf("bgp withdraw mismatch: -(got); +(want): %s", diff)
 		}
 	})
@@ -261,7 +272,7 @@ func TestBgpServer(t *testing.T) {
 				Table:    syscall.RT_TABLE_MAIN,
 			},
 		}
-		if diff := checkRoutes(nlr.getRoutesAdded(), want); diff != "" {
+		if diff := checkRoutes(routeManager.getRoutesAdded(), want); diff != "" {
 			t.Fatalf("bgp add mismatch: -(got); +(want): %s", diff)
 		}
 	})
@@ -289,7 +300,7 @@ func TestBgpServer(t *testing.T) {
 				Table:   syscall.RT_TABLE_MAIN,
 			},
 		}
-		if diff := checkRoutes(nlr.getRoutesFlushed(), want); diff != "" {
+		if diff := checkRoutes(routeManager.getRoutesFlushed(), want); diff != "" {
 			t.Fatalf("bgp flush mismatch: -(got); +(want): %s", diff)
 		}
 	})