From 2e6d8f5293ab7ef305ac94ef53b28d63d6509402 Mon Sep 17 00:00:00 2001
From: Adam Chen <encodeous7@gmail.com>
Date: Mon, 6 Jul 2026 04:37:17 +0000
Subject: [PATCH 1/3] feat: implement futures to better handle async IPC &
 probes

---
 cmd/probe.go             |  49 +++++++-
 core/future.go           |  94 +++++++++++++++
 core/future_test.go      |  52 +++++++++
 core/ipc_handler.go      |  61 ++++++----
 core/ipc_handler_test.go |  40 +++++++
 core/nylon_endpoints.go  | 138 ++++++++++++++++------
 core/nylon_scheduler.go  |  20 ++++
 e2e/probe_test.go        | 203 ++++++++++++++++++++++++++++++++
 integration/ipc_test.go  |  36 ++++++
 protocol/nylon_ipc.pb.go | 243 ++++++++++++++++++++++++++-------------
 protocol/nylon_ipc.proto |  14 ++-
 state/tunables.go        |   2 +
 12 files changed, 810 insertions(+), 142 deletions(-)
 create mode 100644 core/future.go
 create mode 100644 core/future_test.go
 create mode 100644 core/ipc_handler_test.go
 create mode 100644 e2e/probe_test.go

diff --git a/cmd/probe.go b/cmd/probe.go
index 70144c54..22ba9fc7 100644
--- a/cmd/probe.go
+++ b/cmd/probe.go
@@ -3,6 +3,7 @@ package cmd
 import (
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/encodeous/nylon/core"
 	"github.com/encodeous/nylon/protocol"
@@ -17,8 +18,16 @@ var probeCmd = &cobra.Command{
 	Run: func(cmd *cobra.Command, args []string) {
 		itf, _ := cmd.Flags().GetString("interface")
 		jsonOut, _ := cmd.Flags().GetBool("json")
+		timeout, _ := cmd.Flags().GetDuration("timeout")
+		if timeout <= 0 {
+			fmt.Fprintln(os.Stderr, "Error: timeout must be positive")
+			os.Exit(1)
+		}
 		resp, err := core.SendIPCRequest(itf, &protocol.IpcRequest{
-			Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: args[0]}},
+			Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{
+				PeerId:    args[0],
+				TimeoutMs: durationMillis(timeout),
+			}},
 		})
 		if err != nil {
 			fmt.Fprintln(os.Stderr, "Error:", err)
@@ -33,11 +42,14 @@ var probeCmd = &cobra.Command{
 			return
 		}
 		for _, r := range resp.GetProbe().Results {
-			status := "ok"
-			if !r.Success {
-				status = "error: " + r.Error
+			fmt.Printf("  %s  %s", r.Address, endpointProbeStatus(r))
+			if r.Resolved != nil && *r.Resolved != "" && *r.Resolved != r.Address {
+				fmt.Printf("  resolved=%s", *r.Resolved)
 			}
-			fmt.Printf("  %s  %s\n", r.Address, status)
+			if r.Status == protocol.EndpointProbeStatus_ENDPOINT_PROBE_REPLIED {
+				fmt.Printf("  latency=%s", formatDurationNs(r.LatencyNs))
+			}
+			fmt.Println()
 		}
 	},
 }
@@ -46,4 +58,31 @@ func init() {
 	rootCmd.AddCommand(probeCmd)
 	probeCmd.Flags().StringP("interface", "i", "nylon", "Interface name")
 	probeCmd.Flags().Bool("json", false, "Output as JSON")
+	probeCmd.Flags().Duration("timeout", 2*time.Second, "Probe response timeout")
+}
+
+func durationMillis(d time.Duration) uint32 {
+	ms := d.Milliseconds()
+	if ms <= 0 {
+		return 1
+	}
+	if ms > int64(^uint32(0)) {
+		return ^uint32(0)
+	}
+	return uint32(ms)
+}
+
+func endpointProbeStatus(r *protocol.EndpointProbeResult) string {
+	switch r.Status {
+	case protocol.EndpointProbeStatus_ENDPOINT_PROBE_REPLIED:
+		return "ok"
+	case protocol.EndpointProbeStatus_ENDPOINT_PROBE_TIMEOUT:
+		return "timeout"
+	case protocol.EndpointProbeStatus_ENDPOINT_PROBE_SEND_ERROR:
+		return "send-error"
+	case protocol.EndpointProbeStatus_ENDPOINT_PROBE_RESOLVE_ERROR:
+		return "resolve-error"
+	default:
+		return "unknown"
+	}
 }
diff --git a/core/future.go b/core/future.go
new file mode 100644
index 00000000..c08d2ba2
--- /dev/null
+++ b/core/future.go
@@ -0,0 +1,94 @@
+package core
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+var ErrFutureAlreadyAwaited = errors.New("future already awaited")
+
+type futureState[T any] struct {
+	done    chan struct{}
+	once    sync.Once
+	mu      sync.Mutex
+	value   T
+	err     error
+	awaited bool
+}
+
+// Future is a single-assignment async result.
+type Future[T any] struct {
+	state *futureState[T]
+}
+
+// NewFuture returns a Future and its completion function. Only the first
+// completion call is used; later calls are ignored.
+func NewFuture[T any]() (Future[T], func(T, error)) {
+	state := &futureState[T]{done: make(chan struct{})}
+	complete := func(value T, err error) {
+		state.once.Do(func() {
+			state.mu.Lock()
+			state.value = value
+			state.err = err
+			state.mu.Unlock()
+			close(state.done)
+		})
+	}
+	return Future[T]{state: state}, complete
+}
+
+// Await blocks until the future completes or ctx is canceled.
+// A future result can be awaited once; later Await calls return ErrFutureAlreadyAwaited.
+func (f Future[T]) Await(ctx context.Context) (T, error) {
+	select {
+	case <-f.state.done:
+		return f.result()
+	default:
+	}
+	select {
+	case <-f.state.done:
+		return f.result()
+	case <-ctx.Done():
+		var zero T
+		return zero, ctx.Err()
+	}
+}
+
+func (f Future[T]) result() (T, error) {
+	f.state.mu.Lock()
+	defer f.state.mu.Unlock()
+	if f.state.awaited {
+		var zero T
+		return zero, ErrFutureAlreadyAwaited
+	}
+	f.state.awaited = true
+	return f.state.value, f.state.err
+}
+
+// Done returns a receive-only readiness channel for use in select statements.
+// Receive from Done, then call Await to consume the result.
+func (f Future[T]) Done() <-chan struct{} {
+	return f.state.done
+}
+
+// All blocks until every input future completes or ctx is canceled.
+// Values are returned in the same order as the input slice. All awaits and
+// consumes every completed input future in order.
+func All[T any](ctx context.Context, futures []Future[T]) ([]T, error) {
+	values := make([]T, len(futures))
+	var firstErr error
+	for idx, item := range futures {
+		value, err := item.Await(ctx)
+		if err != nil {
+			if ctxErr := ctx.Err(); ctxErr != nil && errors.Is(err, ctxErr) {
+				return nil, ctxErr
+			}
+			if firstErr == nil {
+				firstErr = err
+			}
+		}
+		values[idx] = value
+	}
+	return values, firstErr
+}
diff --git a/core/future_test.go b/core/future_test.go
new file mode 100644
index 00000000..1a25e110
--- /dev/null
+++ b/core/future_test.go
@@ -0,0 +1,52 @@
+package core
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+func TestFutureAwaitOnce(t *testing.T) {
+	future, complete := NewFuture[int]()
+	complete(7, nil)
+
+	value, err := future.Await(context.Background())
+	if err != nil {
+		t.Fatalf("first await failed: %v", err)
+	}
+	if value != 7 {
+		t.Fatalf("expected 7, got %d", value)
+	}
+
+	_, err = future.Await(context.Background())
+	if !errors.Is(err, ErrFutureAlreadyAwaited) {
+		t.Fatalf("expected ErrFutureAlreadyAwaited, got %v", err)
+	}
+}
+
+func TestFutureAllPreservesOrder(t *testing.T) {
+	a, completeA := NewFuture[int]()
+	b, completeB := NewFuture[int]()
+
+	completeB(2, nil)
+	completeA(1, nil)
+
+	values, err := All(context.Background(), []Future[int]{a, b})
+	if err != nil {
+		t.Fatalf("await all failed: %v", err)
+	}
+	if len(values) != 2 || values[0] != 1 || values[1] != 2 {
+		t.Fatalf("unexpected values: %#v", values)
+	}
+}
+
+func TestFutureAllTimeout(t *testing.T) {
+	future, _ := NewFuture[int]()
+	ctx, cancel := context.WithTimeout(context.Background(), 0)
+	defer cancel()
+
+	_, err := All(ctx, []Future[int]{future})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("expected context deadline exceeded, got %v", err)
+	}
+}
diff --git a/core/ipc_handler.go b/core/ipc_handler.go
index f687b24b..7d2bc22f 100644
--- a/core/ipc_handler.go
+++ b/core/ipc_handler.go
@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"cmp"
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"slices"
@@ -19,6 +20,11 @@ import (
 var pjMarshal = protojson.MarshalOptions{EmitUnpopulated: true}
 var pjUnmarshal = protojson.UnmarshalOptions{DiscardUnknown: true}
 
+const (
+	defaultIPCProbeTimeout = 2 * time.Second
+	maxIPCProbeTimeout     = 10 * time.Second
+)
+
 func writeResponse(rw *bufio.ReadWriter, resp *protocol.IpcResponse) error {
 	data, err := pjMarshal.Marshal(resp)
 	if err != nil {
@@ -64,6 +70,13 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 	if _, ok := req.Request.(*protocol.IpcRequest_Trace); ok {
 		return handleTrace(n, rw)
 	}
+	if _, ok := req.Request.(*protocol.IpcRequest_Probe); ok {
+		resp := handleIPCProbe(n, req.GetProbe())
+		if err := writeResponse(rw, resp); err != nil {
+			return err
+		}
+		return device.ErrIPCStatusHandled
+	}
 
 	done := make(chan *protocol.IpcResponse, 1)
 	n.Dispatch(func() error {
@@ -71,8 +84,6 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 		switch req.Request.(type) {
 		case *protocol.IpcRequest_Status:
 			resp = handleStatus(n, req.GetStatus())
-		case *protocol.IpcRequest_Probe:
-			resp = handleIPCProbe(n, req.GetProbe())
 		case *protocol.IpcRequest_Reload:
 			resp = handleIPCReload(n, req.GetReload())
 		default:
@@ -87,7 +98,7 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 	case resp = <-done:
 	case <-n.Context.Done():
 		resp = errResponse("nylon shutting down")
-	case <-time.After(1 * time.Second):
+	case <-time.After(n.IPCDispatchTimeout):
 		// nylon is too busy to handle IPC requests
 		resp = errResponse("timed out waiting for dispatch")
 	}
@@ -210,7 +221,7 @@ func buildEndpoints(neigh *state.Neighbour) []*protocol.EndpointInfo {
 		nep := ep.AsNylonEndpoint()
 		var resolved *string
 		if ap, err := nep.DynEP.Get(); err == nil {
-			resolved = stringPtr(ap.String())
+			resolved = new(ap.String())
 		}
 		eps = append(eps, &protocol.EndpointInfo{
 			Address:         nep.DynEP.Value,
@@ -376,10 +387,6 @@ func keyString(key state.NyPublicKey) string {
 	return string(text)
 }
 
-func stringPtr(v string) *string {
-	return &v
-}
-
 func comparePubRoute(a, b *protocol.PubRoute) int {
 	if c := cmp.Compare(a.Source.Prefix, b.Source.Prefix); c != 0 {
 		return c
@@ -412,20 +419,27 @@ func sortRouteTableEntries(entries []*protocol.RouteTableEntry) {
 }
 
 func handleIPCProbe(n *Nylon, req *protocol.ProbeRequest) *protocol.IpcResponse {
-	neigh := n.RouterState.GetNeighbour(state.NodeId(req.PeerId))
-	if neigh == nil {
-		return errResponse(fmt.Sprintf("peer %q is not a neighbour", req.PeerId))
+	probeTimeout := ipcProbeTimeout(req)
+
+	dispatchCtx, dispatchCancel := context.WithTimeout(n.Context, n.IPCDispatchTimeout)
+	defer dispatchCancel()
+	probes, err := NewDispatchFuture(n, func() ([]Future[*protocol.EndpointProbeResult], error) {
+		return n.sendEndpointProbes(state.NodeId(req.PeerId), probeTimeout)
+	}).Await(dispatchCtx)
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) {
+			return errResponse("timed out waiting for dispatch")
+		}
+		return errResponse(err.Error())
 	}
-	results := make([]*protocol.EndpointProbeResult, 0, len(neigh.Eps))
-	for _, ep := range neigh.Eps {
-		nep := ep.AsNylonEndpoint()
-		addr := nep.DynEP.Value
-		err := n.Probe(neigh.Id, nep, true)
-		r := &protocol.EndpointProbeResult{Address: addr, Success: err == nil}
-		if err != nil {
-			r.Error = err.Error()
+
+	probeCtx, probeCancel := context.WithTimeout(n.Context, probeTimeout+n.IPCDispatchTimeout)
+	defer probeCancel()
+	results, err := All(probeCtx, probes)
+	if err != nil {
+		if ctxErr := probeCtx.Err(); ctxErr != nil && errors.Is(err, ctxErr) {
+			return errResponse(err.Error())
 		}
-		results = append(results, r)
 	}
 	return &protocol.IpcResponse{
 		Ok:       true,
@@ -433,6 +447,13 @@ func handleIPCProbe(n *Nylon, req *protocol.ProbeRequest) *protocol.IpcResponse
 	}
 }
 
+func ipcProbeTimeout(req *protocol.ProbeRequest) time.Duration {
+	if req.TimeoutMs == 0 {
+		return defaultIPCProbeTimeout
+	}
+	return min(time.Duration(req.TimeoutMs)*time.Millisecond, maxIPCProbeTimeout)
+}
+
 func handleIPCReload(n *Nylon, req *protocol.ReloadRequest) *protocol.IpcResponse {
 	data, err := os.ReadFile(n.ConfigPath)
 	if err != nil {
diff --git a/core/ipc_handler_test.go b/core/ipc_handler_test.go
new file mode 100644
index 00000000..8f325891
--- /dev/null
+++ b/core/ipc_handler_test.go
@@ -0,0 +1,40 @@
+package core
+
+import (
+	"testing"
+	"time"
+
+	"github.com/encodeous/nylon/protocol"
+)
+
+func TestIPCProbeTimeout(t *testing.T) {
+	tests := []struct {
+		name      string
+		timeoutMs uint32
+		want      time.Duration
+	}{
+		{
+			name: "default",
+			want: defaultIPCProbeTimeout,
+		},
+		{
+			name:      "user value",
+			timeoutMs: 1500,
+			want:      1500 * time.Millisecond,
+		},
+		{
+			name:      "capped",
+			timeoutMs: uint32((maxIPCProbeTimeout + time.Second) / time.Millisecond),
+			want:      maxIPCProbeTimeout,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ipcProbeTimeout(&protocol.ProbeRequest{TimeoutMs: tt.timeoutMs})
+			if got != tt.want {
+				t.Fatalf("expected %s, got %s", tt.want, got)
+			}
+		})
+	}
+}
diff --git a/core/nylon_endpoints.go b/core/nylon_endpoints.go
index 22b435a5..9e85ea59 100644
--- a/core/nylon_endpoints.go
+++ b/core/nylon_endpoints.go
@@ -1,9 +1,10 @@
 package core
 
 import (
+	"cmp"
+	"fmt"
 	"math/rand/v2"
 	"slices"
-	"sync"
 	"time"
 
 	"github.com/encodeous/nylon/polyamide/conn"
@@ -15,10 +16,63 @@ import (
 
 type EpPing struct {
 	TimeSent time.Time
+	Peer     state.NodeId
+	Complete func(protocol.EndpointProbeStatus, time.Duration)
 }
 
-func (n *Nylon) Probe(node state.NodeId, ep *state.NylonEndpoint, waitErr bool) error {
+func (n *Nylon) sendEndpointProbes(peer state.NodeId, timeout time.Duration) ([]Future[*protocol.EndpointProbeResult], error) {
+	neigh := n.RouterState.GetNeighbour(peer)
+	if neigh == nil {
+		return nil, fmt.Errorf("peer %q is not a neighbour", peer)
+	}
+	eps := slices.Clone(neigh.Eps)
+	slices.SortFunc(eps, func(a, b state.Endpoint) int {
+		return cmp.Compare(a.AsNylonEndpoint().DynEP.Value, b.AsNylonEndpoint().DynEP.Value)
+	})
+	probes := make([]Future[*protocol.EndpointProbeResult], 0, len(eps))
+	for _, ep := range eps {
+		result, _ := n.sendEndpointProbe(neigh.Id, ep.AsNylonEndpoint(), timeout)
+		probes = append(probes, result)
+	}
+	return probes, nil
+}
+
+func (n *Nylon) Probe(node state.NodeId, ep *state.NylonEndpoint) error {
+	_, err := n.sendEndpointProbe(node, ep, 0)
+	return err
+}
+
+func (n *Nylon) sendEndpointProbe(node state.NodeId, ep *state.NylonEndpoint, timeout time.Duration) (Future[*protocol.EndpointProbeResult], error) {
+	address := ep.DynEP.Value
+	resolved := ""
+	resultFuture, completeResult := NewFuture[*protocol.EndpointProbeResult]()
+	completeEndpoint := func(status protocol.EndpointProbeStatus, latency time.Duration, err error) {
+		result := &protocol.EndpointProbeResult{
+			Address:   address,
+			Status:    status,
+			LatencyNs: int64(latency),
+		}
+		if resolved != "" {
+			result.Resolved = new(resolved)
+		}
+		completeResult(result, err)
+	}
+	fail := func(status protocol.EndpointProbeStatus, err error) (Future[*protocol.EndpointProbeResult], error) {
+		completeEndpoint(status, 0, err)
+		return resultFuture, err
+	}
+
+	peer := n.Device.LookupPeer(device.NoisePublicKey(n.GetNode(node).PubKey))
+	if peer == nil {
+		return fail(protocol.EndpointProbeStatus_ENDPOINT_PROBE_SEND_ERROR, fmt.Errorf("wireguard peer %q is not configured", node))
+	}
+	nep, err := ep.GetWgEndpoint(n.Device)
+	if err != nil {
+		return fail(protocol.EndpointProbeStatus_ENDPOINT_PROBE_RESOLVE_ERROR, err)
+	}
+	resolved = nep.DstIPPort().String()
 	token := rand.Uint64()
+	sentAt := time.Now()
 	ping := &protocol.Ny{
 		Type: &protocol.Ny_ProbeOp{
 			ProbeOp: &protocol.Ny_Probe{
@@ -27,33 +81,37 @@ func (n *Nylon) Probe(node state.NodeId, ep *state.NylonEndpoint, waitErr bool)
 			},
 		},
 	}
-	peer := n.Device.LookupPeer(device.NoisePublicKey(n.GetNode(node).PubKey))
-	nep, err := ep.GetWgEndpoint(n.Device)
-	if err != nil {
-		return err
+	var timeoutTimer *time.Timer
+	if timeout > 0 {
+		timeoutTimer = time.AfterFunc(timeout, func() {
+			n.PingBuf.Delete(token)
+			completeEndpoint(protocol.EndpointProbeStatus_ENDPOINT_PROBE_TIMEOUT, 0, nil)
+		})
 	}
 
 	n.PingBuf.Set(token, EpPing{
-		TimeSent: time.Now(),
+		TimeSent: sentAt,
+		Peer:     node,
+		Complete: func(status protocol.EndpointProbeStatus, latency time.Duration) {
+			if timeoutTimer != nil {
+				timeoutTimer.Stop()
+			}
+			completeEndpoint(status, latency, nil)
+		},
 	}, ttlcache.DefaultTTL)
 
-	wg := sync.WaitGroup{}
-	wg.Add(1)
-
-	var sendErr error
 	go func() {
-		defer wg.Done()
-		sendErr = n.SendNylon(ping, nep, peer)
-		if sendErr != nil {
+		err := n.SendNylon(ping, nep, peer)
+		if err != nil {
+			if timeoutTimer != nil {
+				timeoutTimer.Stop()
+			}
 			n.PingBuf.Delete(token)
+			completeEndpoint(protocol.EndpointProbeStatus_ENDPOINT_PROBE_SEND_ERROR, 0, err)
 		}
 	}()
 
-	if waitErr {
-		wg.Wait()
-		return sendErr
-	}
-	return nil
+	return resultFuture, nil
 }
 
 func handleProbe(n *Nylon, pkt *protocol.Ny_Probe, endpoint conn.Endpoint, peer *device.Peer, node state.NodeId) {
@@ -130,28 +188,36 @@ func handleProbePing(n *Nylon, node state.NodeId, wgEndpoint conn.Endpoint) {
 }
 
 func handleProbePong(n *Nylon, node state.NodeId, token uint64, ep conn.Endpoint) {
+	linkHealth, ok := n.PingBuf.GetAndDelete(token)
+	if !ok {
+		return
+	}
+	health := linkHealth.Value()
+	if health.Peer != "" && health.Peer != node {
+		n.Log.Warn("probe came back from wrong peer", "expected", health.Peer, "actual", node)
+		return
+	}
+	receivedAt := time.Now()
+	latency := receivedAt.Sub(health.TimeSent)
+	health.Complete(protocol.EndpointProbeStatus_ENDPOINT_PROBE_REPLIED, latency)
+
 	// check if link exists
 	for _, neigh := range n.RouterState.Neighbours {
 		for _, dep := range neigh.Eps {
 			dpLink := dep.AsNylonEndpoint()
 			ap, err := dpLink.DynEP.Get()
 			if err == nil && ap == ep.DstIPPort() && neigh.Id == node {
-				linkHealth, ok := n.PingBuf.GetAndDelete(token)
-				if ok {
-					health := linkHealth.Value()
-					latency := time.Since(health.TimeSent)
-					// we have a link
-					if n.DBG_log_probe {
-						n.Log.Debug("probe back", "peer", node, "ping", latency)
-					}
-					dpLink.Renew()
-					dpLink.UpdatePing(latency)
-
-					// update wireguard endpoint
-					dpLink.WgEndpoint = ep
-
-					ComputeRoutes(n.RouterState, n)
+				// we have a link
+				if n.DBG_log_probe {
+					n.Log.Debug("probe back", "peer", node, "ping", latency)
 				}
+				dpLink.Renew()
+				dpLink.UpdatePing(latency)
+
+				// update wireguard endpoint
+				dpLink.WgEndpoint = ep
+
+				ComputeRoutes(n.RouterState, n)
 				return
 			}
 		}
@@ -164,7 +230,7 @@ func (n *Nylon) probeLinks(active bool) error {
 	for _, neigh := range n.RouterState.Neighbours {
 		for _, ep := range neigh.Eps {
 			if ep.IsActive() == active {
-				err := n.Probe(neigh.Id, ep.AsNylonEndpoint(), false)
+				err := n.Probe(neigh.Id, ep.AsNylonEndpoint())
 				if err != nil {
 					n.Log.Debug("probe failed", "err", err.Error())
 				}
@@ -202,7 +268,7 @@ func (n *Nylon) probeNew() error {
 				// add the link to the neighbour
 				dpl := state.NewEndpoint(ep, false, nil, &n.RouterTunables)
 				neigh.Eps = append(neigh.Eps, dpl)
-				err := n.Probe(peer, dpl, false)
+				err := n.Probe(peer, dpl)
 				if err != nil {
 					//n.Log.Debug("discovery probe failed", "err", err.Error())
 				}
diff --git a/core/nylon_scheduler.go b/core/nylon_scheduler.go
index c5d53ce9..944c2520 100644
--- a/core/nylon_scheduler.go
+++ b/core/nylon_scheduler.go
@@ -1,12 +1,32 @@
 package core
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"runtime"
 	"time"
 )
 
+func NewDispatchFuture[T any](n *Nylon, fun func() (T, error)) Future[T] {
+	future, complete := NewFuture[T]()
+	dispatch := func() error {
+		value, err := fun()
+		complete(value, err)
+		return nil
+	}
+	select {
+	case n.DispatchChannel <- dispatch:
+	case <-n.Context.Done():
+		var zero T
+		complete(zero, context.Cause(n.Context))
+	default:
+		var zero T
+		complete(zero, fmt.Errorf("dispatch channel is full"))
+	}
+	return future
+}
+
 // Dispatch Dispatches the function to run on the main thread without waiting for it to complete
 func (n *Nylon) Dispatch(fun func() error) {
 	defer func() {
diff --git a/e2e/probe_test.go b/e2e/probe_test.go
new file mode 100644
index 00000000..2b576510
--- /dev/null
+++ b/e2e/probe_test.go
@@ -0,0 +1,203 @@
+//go:build e2e
+
+package e2e
+
+import (
+	"fmt"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/encodeous/nylon/protocol"
+	"github.com/encodeous/nylon/state"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+func TestProbeReportsResolvedDNSAndLatency(t *testing.T) {
+	t.Parallel()
+	h := NewHarness(t)
+
+	dnsIP := GetIP(h.Subnet, 100)
+	node1IP := GetIP(h.Subnet, 2)
+	node2IP := GetIP(h.Subnet, 3)
+	node2Endpoint := "node2.probe.test"
+
+	startProbeDNS(h, dnsIP, fmt.Sprintf(`
+probe.test. 0 IN SOA sns.dns.icann.org. noc.dns.icann.org. 2017042745 7200 3600 1209600 0
+node2.probe.test. 0 IN A %s
+`, node2IP))
+
+	node1Key := state.GenerateKey()
+	node2Key := state.GenerateKey()
+	configDir := h.SetupTestDir()
+	central := state.CentralCfg{
+		Routers: []state.RouterCfg{
+			probeRouter("node1", node1Key.Pubkey(), "10.0.0.1", ""),
+			probeRouter("node2", node2Key.Pubkey(), "10.0.0.2", node2Endpoint),
+		},
+		Graph:     []string{"node1, node2"},
+		Timestamp: time.Now().UnixNano(),
+	}
+	centralPath := h.WriteConfig(configDir, "central.yaml", central)
+
+	node1Cfg := SimpleLocal("node1", node1Key)
+	node1Cfg.DnsResolvers = []string{dnsIP + ":53"}
+	node1Path := h.WriteConfig(configDir, "node1.yaml", node1Cfg)
+	node2Path := h.WriteConfig(configDir, "node2.yaml", SimpleLocal("node2", node2Key))
+
+	h.StartNodes(
+		NodeSpec{Name: "node1", IP: node1IP, CentralConfigPath: centralPath, NodeConfigPath: node1Path},
+		NodeSpec{Name: "node2", IP: node2IP, CentralConfigPath: centralPath, NodeConfigPath: node2Path},
+	)
+	h.WaitForStatus(t, "node1", func(status *protocol.StatusResponse) bool {
+		return HasResolvedEndpoint(status, node2Endpoint, fmt.Sprintf("%s:57175", node2IP))
+	})
+
+	result := waitForProbeResult(t, h, "node1", "node2", node2Endpoint, 2*time.Second, func(result *protocol.EndpointProbeResult) bool {
+		return result.GetStatus() == protocol.EndpointProbeStatus_ENDPOINT_PROBE_REPLIED &&
+			result.GetResolved() == fmt.Sprintf("%s:57175", node2IP) &&
+			result.GetLatencyNs() > 0
+	})
+	t.Logf("probe latency: %s", time.Duration(result.GetLatencyNs()))
+}
+
+func TestProbeReportsResolveErrorForMissingDNS(t *testing.T) {
+	t.Parallel()
+	h := NewHarness(t)
+
+	dnsIP := GetIP(h.Subnet, 100)
+	node1IP := GetIP(h.Subnet, 2)
+	missingEndpoint := "missing.probe.test"
+
+	startProbeDNS(h, dnsIP, `
+probe.test. 0 IN SOA sns.dns.icann.org. noc.dns.icann.org. 2017042745 7200 3600 1209600 0
+`)
+
+	node1Key := state.GenerateKey()
+	node2Key := state.GenerateKey()
+	configDir := h.SetupTestDir()
+	central := state.CentralCfg{
+		Routers: []state.RouterCfg{
+			probeRouter("node1", node1Key.Pubkey(), "10.0.0.1", ""),
+			probeRouter("node2", node2Key.Pubkey(), "10.0.0.2", missingEndpoint),
+		},
+		Graph:     []string{"node1, node2"},
+		Timestamp: time.Now().UnixNano(),
+	}
+	centralPath := h.WriteConfig(configDir, "central.yaml", central)
+
+	node1Cfg := SimpleLocal("node1", node1Key)
+	node1Cfg.DnsResolvers = []string{dnsIP + ":53"}
+	node1Path := h.WriteConfig(configDir, "node1.yaml", node1Cfg)
+
+	h.StartNode("node1", node1IP, centralPath, node1Path)
+
+	waitForProbeResult(t, h, "node1", "node2", missingEndpoint, 200*time.Millisecond, func(result *protocol.EndpointProbeResult) bool {
+		return result.GetStatus() == protocol.EndpointProbeStatus_ENDPOINT_PROBE_RESOLVE_ERROR &&
+			result.GetResolved() == "" &&
+			result.GetLatencyNs() == 0
+	})
+}
+
+func TestProbeReportsTimeout(t *testing.T) {
+	t.Parallel()
+	h := NewHarness(t)
+
+	node1IP := GetIP(h.Subnet, 2)
+	unusedNode2IP := GetIP(h.Subnet, 3)
+	node2Endpoint := fmt.Sprintf("%s:57175", unusedNode2IP)
+
+	node1Key := state.GenerateKey()
+	node2Key := state.GenerateKey()
+	configDir := h.SetupTestDir()
+	central := state.CentralCfg{
+		Routers: []state.RouterCfg{
+			probeRouter("node1", node1Key.Pubkey(), "10.0.0.1", ""),
+			probeRouter("node2", node2Key.Pubkey(), "10.0.0.2", node2Endpoint),
+		},
+		Graph:     []string{"node1, node2"},
+		Timestamp: time.Now().UnixNano(),
+	}
+	centralPath := h.WriteConfig(configDir, "central.yaml", central)
+	node1Path := h.WriteConfig(configDir, "node1.yaml", SimpleLocal("node1", node1Key))
+
+	h.StartNode("node1", node1IP, centralPath, node1Path)
+
+	waitForProbeResult(t, h, "node1", "node2", node2Endpoint, 200*time.Millisecond, func(result *protocol.EndpointProbeResult) bool {
+		return result.GetStatus() == protocol.EndpointProbeStatus_ENDPOINT_PROBE_TIMEOUT &&
+			result.GetLatencyNs() == 0
+	})
+}
+
+func startProbeDNS(h *Harness, dnsIP string, zoneFile string) {
+	corefile := `
+. {
+    file /etc/coredns/probe.test.db probe.test
+    log
+    errors
+}
+`
+	h.StartDNS("dns", dnsIP, corefile, map[string]string{"probe.test.db": zoneFile})
+}
+
+func probeRouter(id string, pubKey state.NyPublicKey, nylonIP string, endpoint string) state.RouterCfg {
+	cfg := state.RouterCfg{
+		NodeCfg: state.NodeCfg{
+			Id:        state.NodeId(id),
+			PubKey:    pubKey,
+			Addresses: []netip.Addr{netip.MustParseAddr(nylonIP)},
+		},
+	}
+	if endpoint != "" {
+		cfg.Endpoints = []*state.DynamicEndpoint{state.NewDynamicEndpoint(endpoint)}
+	}
+	return cfg
+}
+
+func waitForProbeResult(t *testing.T, h *Harness, nodeName, peerName, address string, probeTimeout time.Duration, check func(*protocol.EndpointProbeResult) bool) *protocol.EndpointProbeResult {
+	t.Helper()
+
+	deadline := time.Now().Add(30 * time.Second)
+	last := ""
+	for time.Now().Before(deadline) {
+		resp, stdout, stderr, err := runProbeJSON(h, nodeName, peerName, probeTimeout)
+		if err == nil && resp.GetOk() {
+			for _, result := range resp.GetProbe().GetResults() {
+				if result.GetAddress() == address && check(result) {
+					return result
+				}
+			}
+			last = stdout
+		} else if err != nil {
+			last = fmt.Sprintf("error: %v\nstdout:\n%s\nstderr:\n%s", err, stdout, stderr)
+		} else {
+			last = fmt.Sprintf("probe response not ok: %s\nstdout:\n%s", resp.GetError(), stdout)
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	t.Fatalf("timed out waiting for probe result address=%q from %s to %s. Last result:\n%s", address, nodeName, peerName, last)
+	return nil
+}
+
+func runProbeJSON(h *Harness, nodeName, peerName string, timeout time.Duration) (*protocol.IpcResponse, string, string, error) {
+	stdout, stderr, err := h.Exec(nodeName, []string{
+		"nylon",
+		"probe",
+		"-i",
+		"nylon0",
+		"--timeout",
+		timeout.String(),
+		"--json",
+		peerName,
+	})
+	if err != nil {
+		return nil, stdout, stderr, err
+	}
+
+	resp := &protocol.IpcResponse{}
+	if err := protojson.Unmarshal([]byte(stdout), resp); err != nil {
+		return nil, stdout, stderr, err
+	}
+	return resp, stdout, stderr, nil
+}
diff --git a/integration/ipc_test.go b/integration/ipc_test.go
index d6b86004..fa8ea882 100644
--- a/integration/ipc_test.go
+++ b/integration/ipc_test.go
@@ -96,6 +96,42 @@ func TestIPCStatus(t *testing.T) {
 	assert.GreaterOrEqual(t, len(s.GetFeasibilityDistances()), 1)
 }
 
+func TestIPCProbeReportsTimeout(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	vh := &VirtualHarness{}
+	a1 := "192.168.52.1:1234"
+	b1 := "192.168.52.2:1234"
+	vh.NewNode("a", "10.0.0.1/32")
+	vh.NewNode("b", "10.0.0.2/32")
+	vh.Central.Graph = []string{"a, b"}
+	vh.Endpoints = map[string]state.NodeId{a1: "a", b1: "b"}
+	vh.AddLink(a1, b1).WithPacketLoss(1)
+	vh.AddLink(b1, a1).WithPacketLoss(1)
+	errs := vh.Start()
+	defer vh.Stop()
+
+	time.Sleep(500 * time.Millisecond)
+
+	a := vh.Nylons[vh.IndexOf("a")].Load()
+	resp := ipcCall(t, a, &protocol.IpcRequest{
+		Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 100}},
+	})
+
+	require.True(t, resp.Ok, resp.Error)
+	results := resp.GetProbe().GetResults()
+	require.NotEmpty(t, results)
+	for _, result := range results {
+		assert.Equal(t, protocol.EndpointProbeStatus_ENDPOINT_PROBE_TIMEOUT, result.Status)
+		assert.Zero(t, result.LatencyNs)
+	}
+
+	select {
+	case err := <-errs:
+		t.Fatal(err)
+	default:
+	}
+}
+
 func TestIPCReloadConfig(t *testing.T) {
 	defer goleak.VerifyNone(t)
 	vh, errs := setupTwoNodeHarness(t)
diff --git a/protocol/nylon_ipc.pb.go b/protocol/nylon_ipc.pb.go
index 5be9758f..825f9273 100644
--- a/protocol/nylon_ipc.pb.go
+++ b/protocol/nylon_ipc.pb.go
@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.11
+// 	protoc-gen-go v1.36.6
 // 	protoc        v3.21.12
 // source: protocol/nylon_ipc.proto
 
@@ -73,6 +73,61 @@ func (ReloadResult) EnumDescriptor() ([]byte, []int) {
 	return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{0}
 }
 
+type EndpointProbeStatus int32
+
+const (
+	EndpointProbeStatus_ENDPOINT_PROBE_STATUS_UNSPECIFIED EndpointProbeStatus = 0
+	EndpointProbeStatus_ENDPOINT_PROBE_REPLIED            EndpointProbeStatus = 1
+	EndpointProbeStatus_ENDPOINT_PROBE_TIMEOUT            EndpointProbeStatus = 2
+	EndpointProbeStatus_ENDPOINT_PROBE_SEND_ERROR         EndpointProbeStatus = 3
+	EndpointProbeStatus_ENDPOINT_PROBE_RESOLVE_ERROR      EndpointProbeStatus = 4
+)
+
+// Enum value maps for EndpointProbeStatus.
+var (
+	EndpointProbeStatus_name = map[int32]string{
+		0: "ENDPOINT_PROBE_STATUS_UNSPECIFIED",
+		1: "ENDPOINT_PROBE_REPLIED",
+		2: "ENDPOINT_PROBE_TIMEOUT",
+		3: "ENDPOINT_PROBE_SEND_ERROR",
+		4: "ENDPOINT_PROBE_RESOLVE_ERROR",
+	}
+	EndpointProbeStatus_value = map[string]int32{
+		"ENDPOINT_PROBE_STATUS_UNSPECIFIED": 0,
+		"ENDPOINT_PROBE_REPLIED":            1,
+		"ENDPOINT_PROBE_TIMEOUT":            2,
+		"ENDPOINT_PROBE_SEND_ERROR":         3,
+		"ENDPOINT_PROBE_RESOLVE_ERROR":      4,
+	}
+)
+
+func (x EndpointProbeStatus) Enum() *EndpointProbeStatus {
+	p := new(EndpointProbeStatus)
+	*p = x
+	return p
+}
+
+func (x EndpointProbeStatus) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (EndpointProbeStatus) Descriptor() protoreflect.EnumDescriptor {
+	return file_protocol_nylon_ipc_proto_enumTypes[1].Descriptor()
+}
+
+func (EndpointProbeStatus) Type() protoreflect.EnumType {
+	return &file_protocol_nylon_ipc_proto_enumTypes[1]
+}
+
+func (x EndpointProbeStatus) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use EndpointProbeStatus.Descriptor instead.
+func (EndpointProbeStatus) EnumDescriptor() ([]byte, []int) {
+	return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{1}
+}
+
 type StatusRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	unknownFields protoimpl.UnknownFields
@@ -112,6 +167,7 @@ func (*StatusRequest) Descriptor() ([]byte, []int) {
 type ProbeRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	PeerId        string                 `protobuf:"bytes,1,opt,name=peer_id,json=peerId,proto3" json:"peer_id,omitempty"`
+	TimeoutMs     uint32                 `protobuf:"varint,2,opt,name=timeout_ms,json=timeoutMs,proto3" json:"timeout_ms,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -153,6 +209,13 @@ func (x *ProbeRequest) GetPeerId() string {
 	return ""
 }
 
+func (x *ProbeRequest) GetTimeoutMs() uint32 {
+	if x != nil {
+		return x.TimeoutMs
+	}
+	return 0
+}
+
 type ReloadRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	unknownFields protoimpl.UnknownFields
@@ -1324,8 +1387,9 @@ func (x *StatusResponse) GetFeasibilityDistances() []*FeasibilityDistance {
 type EndpointProbeResult struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Address       string                 `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
-	Success       bool                   `protobuf:"varint,2,opt,name=success,proto3" json:"success,omitempty"`
-	Error         string                 `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"`
+	Resolved      *string                `protobuf:"bytes,2,opt,name=resolved,proto3,oneof" json:"resolved,omitempty"`
+	Status        EndpointProbeStatus    `protobuf:"varint,3,opt,name=status,proto3,enum=proto.EndpointProbeStatus" json:"status,omitempty"`
+	LatencyNs     int64                  `protobuf:"varint,4,opt,name=latency_ns,json=latencyNs,proto3" json:"latency_ns,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1367,18 +1431,25 @@ func (x *EndpointProbeResult) GetAddress() string {
 	return ""
 }
 
-func (x *EndpointProbeResult) GetSuccess() bool {
+func (x *EndpointProbeResult) GetResolved() string {
+	if x != nil && x.Resolved != nil {
+		return *x.Resolved
+	}
+	return ""
+}
+
+func (x *EndpointProbeResult) GetStatus() EndpointProbeStatus {
 	if x != nil {
-		return x.Success
+		return x.Status
 	}
-	return false
+	return EndpointProbeStatus_ENDPOINT_PROBE_STATUS_UNSPECIFIED
 }
 
-func (x *EndpointProbeResult) GetError() string {
+func (x *EndpointProbeResult) GetLatencyNs() int64 {
 	if x != nil {
-		return x.Error
+		return x.LatencyNs
 	}
-	return ""
+	return 0
 }
 
 type ProbeResponse struct {
@@ -1770,9 +1841,11 @@ var File_protocol_nylon_ipc_proto protoreflect.FileDescriptor
 const file_protocol_nylon_ipc_proto_rawDesc = "" +
 	"\n" +
 	"\x18protocol/nylon_ipc.proto\x12\x05proto\"\x0f\n" +
-	"\rStatusRequest\"'\n" +
+	"\rStatusRequest\"F\n" +
 	"\fProbeRequest\x12\x17\n" +
-	"\apeer_id\x18\x01 \x01(\tR\x06peerId\"\x0f\n" +
+	"\apeer_id\x18\x01 \x01(\tR\x06peerId\x12\x1d\n" +
+	"\n" +
+	"timeout_ms\x18\x02 \x01(\rR\ttimeoutMs\"\x0f\n" +
 	"\rReloadRequest\"\x0e\n" +
 	"\fTraceRequest\"9\n" +
 	"\x06Source\x12\x17\n" +
@@ -1871,11 +1944,14 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" +
 	"neighbours\x18\x02 \x03(\v2\x14.proto.NeighbourInfoR\n" +
 	"neighbours\x12*\n" +
 	"\x06routes\x18\x03 \x01(\v2\x12.proto.RouteTablesR\x06routes\x12O\n" +
-	"\x15feasibility_distances\x18\x04 \x03(\v2\x1a.proto.FeasibilityDistanceR\x14feasibilityDistances\"_\n" +
+	"\x15feasibility_distances\x18\x04 \x03(\v2\x1a.proto.FeasibilityDistanceR\x14feasibilityDistances\"\xb0\x01\n" +
 	"\x13EndpointProbeResult\x12\x18\n" +
-	"\aaddress\x18\x01 \x01(\tR\aaddress\x12\x18\n" +
-	"\asuccess\x18\x02 \x01(\bR\asuccess\x12\x14\n" +
-	"\x05error\x18\x03 \x01(\tR\x05error\"E\n" +
+	"\aaddress\x18\x01 \x01(\tR\aaddress\x12\x1f\n" +
+	"\bresolved\x18\x02 \x01(\tH\x00R\bresolved\x88\x01\x01\x122\n" +
+	"\x06status\x18\x03 \x01(\x0e2\x1a.proto.EndpointProbeStatusR\x06status\x12\x1d\n" +
+	"\n" +
+	"latency_ns\x18\x04 \x01(\x03R\tlatencyNsB\v\n" +
+	"\t_resolved\"E\n" +
 	"\rProbeResponse\x124\n" +
 	"\aresults\x18\x01 \x03(\v2\x1a.proto.EndpointProbeResultR\aresults\"W\n" +
 	"\x0eReloadResponse\x12+\n" +
@@ -1904,7 +1980,13 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" +
 	"\x04NOOP\x10\x00\x12\v\n" +
 	"\aAPPLIED\x10\x01\x12\f\n" +
 	"\bREJECTED\x10\x02\x12\x14\n" +
-	"\x10RESTART_REQUIRED\x10\x03B\vZ\tprotocol/b\x06proto3"
+	"\x10RESTART_REQUIRED\x10\x03*\xb5\x01\n" +
+	"\x13EndpointProbeStatus\x12%\n" +
+	"!ENDPOINT_PROBE_STATUS_UNSPECIFIED\x10\x00\x12\x1a\n" +
+	"\x16ENDPOINT_PROBE_REPLIED\x10\x01\x12\x1a\n" +
+	"\x16ENDPOINT_PROBE_TIMEOUT\x10\x02\x12\x1d\n" +
+	"\x19ENDPOINT_PROBE_SEND_ERROR\x10\x03\x12 \n" +
+	"\x1cENDPOINT_PROBE_RESOLVE_ERROR\x10\x04B\vZ\tprotocol/b\x06proto3"
 
 var (
 	file_protocol_nylon_ipc_proto_rawDescOnce sync.Once
@@ -1918,73 +2000,75 @@ func file_protocol_nylon_ipc_proto_rawDescGZIP() []byte {
 	return file_protocol_nylon_ipc_proto_rawDescData
 }
 
-var file_protocol_nylon_ipc_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_protocol_nylon_ipc_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
 var file_protocol_nylon_ipc_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
 var file_protocol_nylon_ipc_proto_goTypes = []any{
 	(ReloadResult)(0),           // 0: proto.ReloadResult
-	(*StatusRequest)(nil),       // 1: proto.StatusRequest
-	(*ProbeRequest)(nil),        // 2: proto.ProbeRequest
-	(*ReloadRequest)(nil),       // 3: proto.ReloadRequest
-	(*TraceRequest)(nil),        // 4: proto.TraceRequest
-	(*Source)(nil),              // 5: proto.Source
-	(*FD)(nil),                  // 6: proto.FD
-	(*PubRoute)(nil),            // 7: proto.PubRoute
-	(*NeighRoute)(nil),          // 8: proto.NeighRoute
-	(*SelRoute)(nil),            // 9: proto.SelRoute
-	(*Advertisement)(nil),       // 10: proto.Advertisement
-	(*EndpointInfo)(nil),        // 11: proto.EndpointInfo
-	(*WireGuardPeerStats)(nil),  // 12: proto.WireGuardPeerStats
-	(*NeighbourInfo)(nil),       // 13: proto.NeighbourInfo
-	(*RouteTableEntry)(nil),     // 14: proto.RouteTableEntry
-	(*RouteTables)(nil),         // 15: proto.RouteTables
-	(*SeqnoEntry)(nil),          // 16: proto.SeqnoEntry
-	(*FeasibilityDistance)(nil), // 17: proto.FeasibilityDistance
-	(*NodeStats)(nil),           // 18: proto.NodeStats
-	(*NodeStatus)(nil),          // 19: proto.NodeStatus
-	(*StatusResponse)(nil),      // 20: proto.StatusResponse
-	(*EndpointProbeResult)(nil), // 21: proto.EndpointProbeResult
-	(*ProbeResponse)(nil),       // 22: proto.ProbeResponse
-	(*ReloadResponse)(nil),      // 23: proto.ReloadResponse
-	(*TraceEvent)(nil),          // 24: proto.TraceEvent
-	(*IpcRequest)(nil),          // 25: proto.IpcRequest
-	(*IpcResponse)(nil),         // 26: proto.IpcResponse
+	(EndpointProbeStatus)(0),    // 1: proto.EndpointProbeStatus
+	(*StatusRequest)(nil),       // 2: proto.StatusRequest
+	(*ProbeRequest)(nil),        // 3: proto.ProbeRequest
+	(*ReloadRequest)(nil),       // 4: proto.ReloadRequest
+	(*TraceRequest)(nil),        // 5: proto.TraceRequest
+	(*Source)(nil),              // 6: proto.Source
+	(*FD)(nil),                  // 7: proto.FD
+	(*PubRoute)(nil),            // 8: proto.PubRoute
+	(*NeighRoute)(nil),          // 9: proto.NeighRoute
+	(*SelRoute)(nil),            // 10: proto.SelRoute
+	(*Advertisement)(nil),       // 11: proto.Advertisement
+	(*EndpointInfo)(nil),        // 12: proto.EndpointInfo
+	(*WireGuardPeerStats)(nil),  // 13: proto.WireGuardPeerStats
+	(*NeighbourInfo)(nil),       // 14: proto.NeighbourInfo
+	(*RouteTableEntry)(nil),     // 15: proto.RouteTableEntry
+	(*RouteTables)(nil),         // 16: proto.RouteTables
+	(*SeqnoEntry)(nil),          // 17: proto.SeqnoEntry
+	(*FeasibilityDistance)(nil), // 18: proto.FeasibilityDistance
+	(*NodeStats)(nil),           // 19: proto.NodeStats
+	(*NodeStatus)(nil),          // 20: proto.NodeStatus
+	(*StatusResponse)(nil),      // 21: proto.StatusResponse
+	(*EndpointProbeResult)(nil), // 22: proto.EndpointProbeResult
+	(*ProbeResponse)(nil),       // 23: proto.ProbeResponse
+	(*ReloadResponse)(nil),      // 24: proto.ReloadResponse
+	(*TraceEvent)(nil),          // 25: proto.TraceEvent
+	(*IpcRequest)(nil),          // 26: proto.IpcRequest
+	(*IpcResponse)(nil),         // 27: proto.IpcResponse
 }
 var file_protocol_nylon_ipc_proto_depIdxs = []int32{
-	5,  // 0: proto.PubRoute.source:type_name -> proto.Source
-	6,  // 1: proto.PubRoute.fd:type_name -> proto.FD
-	7,  // 2: proto.NeighRoute.pub_route:type_name -> proto.PubRoute
-	7,  // 3: proto.SelRoute.pub_route:type_name -> proto.PubRoute
-	11, // 4: proto.NeighbourInfo.endpoints:type_name -> proto.EndpointInfo
-	8,  // 5: proto.NeighbourInfo.routes:type_name -> proto.NeighRoute
-	10, // 6: proto.NeighbourInfo.advertised:type_name -> proto.Advertisement
-	12, // 7: proto.NeighbourInfo.wireguard:type_name -> proto.WireGuardPeerStats
-	9,  // 8: proto.RouteTables.selected:type_name -> proto.SelRoute
-	14, // 9: proto.RouteTables.forward:type_name -> proto.RouteTableEntry
-	14, // 10: proto.RouteTables.exit:type_name -> proto.RouteTableEntry
-	5,  // 11: proto.FeasibilityDistance.source:type_name -> proto.Source
-	6,  // 12: proto.FeasibilityDistance.fd:type_name -> proto.FD
-	10, // 13: proto.NodeStatus.advertised:type_name -> proto.Advertisement
-	16, // 14: proto.NodeStatus.seqnos:type_name -> proto.SeqnoEntry
-	18, // 15: proto.NodeStatus.stats:type_name -> proto.NodeStats
-	19, // 16: proto.StatusResponse.node:type_name -> proto.NodeStatus
-	13, // 17: proto.StatusResponse.neighbours:type_name -> proto.NeighbourInfo
-	15, // 18: proto.StatusResponse.routes:type_name -> proto.RouteTables
-	17, // 19: proto.StatusResponse.feasibility_distances:type_name -> proto.FeasibilityDistance
-	21, // 20: proto.ProbeResponse.results:type_name -> proto.EndpointProbeResult
-	0,  // 21: proto.ReloadResponse.result:type_name -> proto.ReloadResult
-	1,  // 22: proto.IpcRequest.status:type_name -> proto.StatusRequest
-	2,  // 23: proto.IpcRequest.probe:type_name -> proto.ProbeRequest
-	3,  // 24: proto.IpcRequest.reload:type_name -> proto.ReloadRequest
-	4,  // 25: proto.IpcRequest.trace:type_name -> proto.TraceRequest
-	20, // 26: proto.IpcResponse.status:type_name -> proto.StatusResponse
-	22, // 27: proto.IpcResponse.probe:type_name -> proto.ProbeResponse
-	23, // 28: proto.IpcResponse.reload:type_name -> proto.ReloadResponse
-	24, // 29: proto.IpcResponse.trace:type_name -> proto.TraceEvent
-	30, // [30:30] is the sub-list for method output_type
-	30, // [30:30] is the sub-list for method input_type
-	30, // [30:30] is the sub-list for extension type_name
-	30, // [30:30] is the sub-list for extension extendee
-	0,  // [0:30] is the sub-list for field type_name
+	6,  // 0: proto.PubRoute.source:type_name -> proto.Source
+	7,  // 1: proto.PubRoute.fd:type_name -> proto.FD
+	8,  // 2: proto.NeighRoute.pub_route:type_name -> proto.PubRoute
+	8,  // 3: proto.SelRoute.pub_route:type_name -> proto.PubRoute
+	12, // 4: proto.NeighbourInfo.endpoints:type_name -> proto.EndpointInfo
+	9,  // 5: proto.NeighbourInfo.routes:type_name -> proto.NeighRoute
+	11, // 6: proto.NeighbourInfo.advertised:type_name -> proto.Advertisement
+	13, // 7: proto.NeighbourInfo.wireguard:type_name -> proto.WireGuardPeerStats
+	10, // 8: proto.RouteTables.selected:type_name -> proto.SelRoute
+	15, // 9: proto.RouteTables.forward:type_name -> proto.RouteTableEntry
+	15, // 10: proto.RouteTables.exit:type_name -> proto.RouteTableEntry
+	6,  // 11: proto.FeasibilityDistance.source:type_name -> proto.Source
+	7,  // 12: proto.FeasibilityDistance.fd:type_name -> proto.FD
+	11, // 13: proto.NodeStatus.advertised:type_name -> proto.Advertisement
+	17, // 14: proto.NodeStatus.seqnos:type_name -> proto.SeqnoEntry
+	19, // 15: proto.NodeStatus.stats:type_name -> proto.NodeStats
+	20, // 16: proto.StatusResponse.node:type_name -> proto.NodeStatus
+	14, // 17: proto.StatusResponse.neighbours:type_name -> proto.NeighbourInfo
+	16, // 18: proto.StatusResponse.routes:type_name -> proto.RouteTables
+	18, // 19: proto.StatusResponse.feasibility_distances:type_name -> proto.FeasibilityDistance
+	1,  // 20: proto.EndpointProbeResult.status:type_name -> proto.EndpointProbeStatus
+	22, // 21: proto.ProbeResponse.results:type_name -> proto.EndpointProbeResult
+	0,  // 22: proto.ReloadResponse.result:type_name -> proto.ReloadResult
+	2,  // 23: proto.IpcRequest.status:type_name -> proto.StatusRequest
+	3,  // 24: proto.IpcRequest.probe:type_name -> proto.ProbeRequest
+	4,  // 25: proto.IpcRequest.reload:type_name -> proto.ReloadRequest
+	5,  // 26: proto.IpcRequest.trace:type_name -> proto.TraceRequest
+	21, // 27: proto.IpcResponse.status:type_name -> proto.StatusResponse
+	23, // 28: proto.IpcResponse.probe:type_name -> proto.ProbeResponse
+	24, // 29: proto.IpcResponse.reload:type_name -> proto.ReloadResponse
+	25, // 30: proto.IpcResponse.trace:type_name -> proto.TraceEvent
+	31, // [31:31] is the sub-list for method output_type
+	31, // [31:31] is the sub-list for method input_type
+	31, // [31:31] is the sub-list for extension type_name
+	31, // [31:31] is the sub-list for extension extendee
+	0,  // [0:31] is the sub-list for field type_name
 }
 
 func init() { file_protocol_nylon_ipc_proto_init() }
@@ -1994,6 +2078,7 @@ func file_protocol_nylon_ipc_proto_init() {
 	}
 	file_protocol_nylon_ipc_proto_msgTypes[10].OneofWrappers = []any{}
 	file_protocol_nylon_ipc_proto_msgTypes[11].OneofWrappers = []any{}
+	file_protocol_nylon_ipc_proto_msgTypes[20].OneofWrappers = []any{}
 	file_protocol_nylon_ipc_proto_msgTypes[24].OneofWrappers = []any{
 		(*IpcRequest_Status)(nil),
 		(*IpcRequest_Probe)(nil),
@@ -2011,7 +2096,7 @@ func file_protocol_nylon_ipc_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_protocol_nylon_ipc_proto_rawDesc), len(file_protocol_nylon_ipc_proto_rawDesc)),
-			NumEnums:      1,
+			NumEnums:      2,
 			NumMessages:   26,
 			NumExtensions: 0,
 			NumServices:   0,
diff --git a/protocol/nylon_ipc.proto b/protocol/nylon_ipc.proto
index 080276ca..8cad7484 100644
--- a/protocol/nylon_ipc.proto
+++ b/protocol/nylon_ipc.proto
@@ -14,6 +14,7 @@ message StatusRequest {}
 
 message ProbeRequest {
   string peer_id = 1;
+  uint32 timeout_ms = 2;
 }
 
 message ReloadRequest {}
@@ -133,10 +134,19 @@ message StatusResponse {
   repeated FeasibilityDistance feasibility_distances = 4;
 }
 
+enum EndpointProbeStatus {
+  ENDPOINT_PROBE_STATUS_UNSPECIFIED = 0;
+  ENDPOINT_PROBE_REPLIED = 1;
+  ENDPOINT_PROBE_TIMEOUT = 2;
+  ENDPOINT_PROBE_SEND_ERROR = 3;
+  ENDPOINT_PROBE_RESOLVE_ERROR = 4;
+}
+
 message EndpointProbeResult {
   string address = 1;
-  bool success = 2;
-  string error = 3;
+  optional string resolved = 2;
+  EndpointProbeStatus status = 3;
+  int64 latency_ns = 4;
 }
 
 message ProbeResponse {
diff --git a/state/tunables.go b/state/tunables.go
index 2c5f7b8a..30a70000 100644
--- a/state/tunables.go
+++ b/state/tunables.go
@@ -15,6 +15,7 @@ type RouterTunables struct {
 	StarvationDelay       time.Duration
 	SeqnoDedupTTL         time.Duration
 	NeighbourIOFlushDelay time.Duration
+	IPCDispatchTimeout    time.Duration
 	SafeMTU               int
 
 	// WindowSamples is the sliding window size
@@ -70,6 +71,7 @@ func DefaultRouterTunables() RouterTunables {
 		StarvationDelay:       time.Millisecond * 100,
 		SeqnoDedupTTL:         time.Second * 3,
 		NeighbourIOFlushDelay: time.Millisecond * 500,
+		IPCDispatchTimeout:    time.Second,
 		SafeMTU:               1200,
 
 		WindowSamples:           int((time.Second * 60) / probeDelay),

From 16f18bcc99c84b055a09917574ebdba8fac3248b Mon Sep 17 00:00:00 2001
From: Adam Chen <encodeous7@gmail.com>
Date: Mon, 6 Jul 2026 12:19:10 +0000
Subject: [PATCH 2/3] refactor: rename All to AwaitAll

---
 core/future.go      | 6 +++---
 core/future_test.go | 4 ++--
 core/ipc_handler.go | 2 +-
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/core/future.go b/core/future.go
index c08d2ba2..2135729d 100644
--- a/core/future.go
+++ b/core/future.go
@@ -72,10 +72,10 @@ func (f Future[T]) Done() <-chan struct{} {
 	return f.state.done
 }
 
-// All blocks until every input future completes or ctx is canceled.
-// Values are returned in the same order as the input slice. All awaits and
+// AwaitAll blocks until every input future completes or ctx is canceled.
+// Values are returned in the same order as the input slice. AwaitAll awaits and
 // consumes every completed input future in order.
-func All[T any](ctx context.Context, futures []Future[T]) ([]T, error) {
+func AwaitAll[T any](ctx context.Context, futures []Future[T]) ([]T, error) {
 	values := make([]T, len(futures))
 	var firstErr error
 	for idx, item := range futures {
diff --git a/core/future_test.go b/core/future_test.go
index 1a25e110..281b4270 100644
--- a/core/future_test.go
+++ b/core/future_test.go
@@ -31,7 +31,7 @@ func TestFutureAllPreservesOrder(t *testing.T) {
 	completeB(2, nil)
 	completeA(1, nil)
 
-	values, err := All(context.Background(), []Future[int]{a, b})
+	values, err := AwaitAll(context.Background(), []Future[int]{a, b})
 	if err != nil {
 		t.Fatalf("await all failed: %v", err)
 	}
@@ -45,7 +45,7 @@ func TestFutureAllTimeout(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 0)
 	defer cancel()
 
-	_, err := All(ctx, []Future[int]{future})
+	_, err := AwaitAll(ctx, []Future[int]{future})
 	if !errors.Is(err, context.DeadlineExceeded) {
 		t.Fatalf("expected context deadline exceeded, got %v", err)
 	}
diff --git a/core/ipc_handler.go b/core/ipc_handler.go
index 7d2bc22f..01078bd5 100644
--- a/core/ipc_handler.go
+++ b/core/ipc_handler.go
@@ -435,7 +435,7 @@ func handleIPCProbe(n *Nylon, req *protocol.ProbeRequest) *protocol.IpcResponse
 
 	probeCtx, probeCancel := context.WithTimeout(n.Context, probeTimeout+n.IPCDispatchTimeout)
 	defer probeCancel()
-	results, err := All(probeCtx, probes)
+	results, err := AwaitAll(probeCtx, probes)
 	if err != nil {
 		if ctxErr := probeCtx.Err(); ctxErr != nil && errors.Is(err, ctxErr) {
 			return errResponse(err.Error())

From 4a9acfe761d446b8d80431d4c2486301fee088e4 Mon Sep 17 00:00:00 2001
From: Adam Chen <encodeous7@gmail.com>
Date: Tue, 7 Jul 2026 00:53:48 +0000
Subject: [PATCH 3/3] fix: small fixes

---
 core/future.go           | 48 +++++++++++++++++++--------
 core/future_test.go      | 70 ++++++++++++++++++++++++++++++++++++++++
 protocol/nylon_ipc.pb.go | 14 ++++----
 protocol/nylon_ipc.proto |  6 ++--
 4 files changed, 114 insertions(+), 24 deletions(-)

diff --git a/core/future.go b/core/future.go
index 2135729d..7603fd21 100644
--- a/core/future.go
+++ b/core/future.go
@@ -9,12 +9,13 @@ import (
 var ErrFutureAlreadyAwaited = errors.New("future already awaited")
 
 type futureState[T any] struct {
-	done    chan struct{}
-	once    sync.Once
-	mu      sync.Mutex
-	value   T
-	err     error
-	awaited bool
+	done     chan struct{}
+	once     sync.Once
+	mu       sync.Mutex
+	value    T
+	err      error
+	awaiting bool
+	awaited  bool
 }
 
 // Future is a single-assignment async result.
@@ -38,34 +39,53 @@ func NewFuture[T any]() (Future[T], func(T, error)) {
 	return Future[T]{state: state}, complete
 }
 
-// Await blocks until the future completes or ctx is canceled.
-// A future result can be awaited once; later Await calls return ErrFutureAlreadyAwaited.
+// Await blocks until the future completes or ctx is canceled. Only one Await
+// call may wait for or consume a result at a time; concurrent calls and calls
+// after the result has been consumed return ErrFutureAlreadyAwaited.
 func (f Future[T]) Await(ctx context.Context) (T, error) {
+	if err := f.beginAwait(); err != nil {
+		var zero T
+		return zero, err
+	}
 	select {
 	case <-f.state.done:
-		return f.result()
+		return f.finishAwait()
 	default:
 	}
 	select {
 	case <-f.state.done:
-		return f.result()
+		return f.finishAwait()
 	case <-ctx.Done():
+		f.cancelAwait()
 		var zero T
 		return zero, ctx.Err()
 	}
 }
 
-func (f Future[T]) result() (T, error) {
+func (f Future[T]) beginAwait() error {
 	f.state.mu.Lock()
 	defer f.state.mu.Unlock()
-	if f.state.awaited {
-		var zero T
-		return zero, ErrFutureAlreadyAwaited
+	if f.state.awaited || f.state.awaiting {
+		return ErrFutureAlreadyAwaited
 	}
+	f.state.awaiting = true
+	return nil
+}
+
+func (f Future[T]) finishAwait() (T, error) {
+	f.state.mu.Lock()
+	defer f.state.mu.Unlock()
+	f.state.awaiting = false
 	f.state.awaited = true
 	return f.state.value, f.state.err
 }
 
+func (f Future[T]) cancelAwait() {
+	f.state.mu.Lock()
+	defer f.state.mu.Unlock()
+	f.state.awaiting = false
+}
+
 // Done returns a receive-only readiness channel for use in select statements.
 // Receive from Done, then call Await to consume the result.
 func (f Future[T]) Done() <-chan struct{} {
diff --git a/core/future_test.go b/core/future_test.go
index 281b4270..52d9aa3e 100644
--- a/core/future_test.go
+++ b/core/future_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 )
 
 func TestFutureAwaitOnce(t *testing.T) {
@@ -24,6 +25,60 @@ func TestFutureAwaitOnce(t *testing.T) {
 	}
 }
 
+func TestFutureAwaitRejectsConcurrentWaiter(t *testing.T) {
+	future, complete := NewFuture[int]()
+	firstValue := make(chan int, 1)
+	firstErr := make(chan error, 1)
+
+	go func() {
+		value, err := future.Await(context.Background())
+		if err != nil {
+			firstErr <- err
+			return
+		}
+		firstValue <- value
+	}()
+
+	waitForFutureAwaiting(t, future)
+
+	_, err := future.Await(context.Background())
+	if !errors.Is(err, ErrFutureAlreadyAwaited) {
+		t.Fatalf("expected ErrFutureAlreadyAwaited, got %v", err)
+	}
+
+	complete(9, nil)
+	select {
+	case err := <-firstErr:
+		t.Fatalf("first await failed: %v", err)
+	case value := <-firstValue:
+		if value != 9 {
+			t.Fatalf("expected first await to get 9, got %d", value)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for first await")
+	}
+}
+
+func TestFutureAwaitCancelDoesNotConsumeResult(t *testing.T) {
+	future, complete := NewFuture[int]()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := future.Await(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context canceled, got %v", err)
+	}
+
+	complete(11, nil)
+	value, err := future.Await(context.Background())
+	if err != nil {
+		t.Fatalf("second await failed: %v", err)
+	}
+	if value != 11 {
+		t.Fatalf("expected 11, got %d", value)
+	}
+}
+
 func TestFutureAllPreservesOrder(t *testing.T) {
 	a, completeA := NewFuture[int]()
 	b, completeB := NewFuture[int]()
@@ -50,3 +105,18 @@ func TestFutureAllTimeout(t *testing.T) {
 		t.Fatalf("expected context deadline exceeded, got %v", err)
 	}
 }
+
+func waitForFutureAwaiting[T any](t *testing.T, future Future[T]) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		future.state.mu.Lock()
+		awaiting := future.state.awaiting
+		future.state.mu.Unlock()
+		if awaiting {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatal("timed out waiting for future to have a waiter")
+}
diff --git a/protocol/nylon_ipc.pb.go b/protocol/nylon_ipc.pb.go
index 825f9273..16bcf0b9 100644
--- a/protocol/nylon_ipc.pb.go
+++ b/protocol/nylon_ipc.pb.go
@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.36.6
+// 	protoc-gen-go v1.36.11
 // 	protoc        v3.21.12
 // source: protocol/nylon_ipc.proto
 
@@ -1387,9 +1387,9 @@ func (x *StatusResponse) GetFeasibilityDistances() []*FeasibilityDistance {
 type EndpointProbeResult struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Address       string                 `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
-	Resolved      *string                `protobuf:"bytes,2,opt,name=resolved,proto3,oneof" json:"resolved,omitempty"`
-	Status        EndpointProbeStatus    `protobuf:"varint,3,opt,name=status,proto3,enum=proto.EndpointProbeStatus" json:"status,omitempty"`
-	LatencyNs     int64                  `protobuf:"varint,4,opt,name=latency_ns,json=latencyNs,proto3" json:"latency_ns,omitempty"`
+	Resolved      *string                `protobuf:"bytes,4,opt,name=resolved,proto3,oneof" json:"resolved,omitempty"`
+	Status        EndpointProbeStatus    `protobuf:"varint,5,opt,name=status,proto3,enum=proto.EndpointProbeStatus" json:"status,omitempty"`
+	LatencyNs     int64                  `protobuf:"varint,6,opt,name=latency_ns,json=latencyNs,proto3" json:"latency_ns,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1947,10 +1947,10 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" +
 	"\x15feasibility_distances\x18\x04 \x03(\v2\x1a.proto.FeasibilityDistanceR\x14feasibilityDistances\"\xb0\x01\n" +
 	"\x13EndpointProbeResult\x12\x18\n" +
 	"\aaddress\x18\x01 \x01(\tR\aaddress\x12\x1f\n" +
-	"\bresolved\x18\x02 \x01(\tH\x00R\bresolved\x88\x01\x01\x122\n" +
-	"\x06status\x18\x03 \x01(\x0e2\x1a.proto.EndpointProbeStatusR\x06status\x12\x1d\n" +
+	"\bresolved\x18\x04 \x01(\tH\x00R\bresolved\x88\x01\x01\x122\n" +
+	"\x06status\x18\x05 \x01(\x0e2\x1a.proto.EndpointProbeStatusR\x06status\x12\x1d\n" +
 	"\n" +
-	"latency_ns\x18\x04 \x01(\x03R\tlatencyNsB\v\n" +
+	"latency_ns\x18\x06 \x01(\x03R\tlatencyNsB\v\n" +
 	"\t_resolved\"E\n" +
 	"\rProbeResponse\x124\n" +
 	"\aresults\x18\x01 \x03(\v2\x1a.proto.EndpointProbeResultR\aresults\"W\n" +
diff --git a/protocol/nylon_ipc.proto b/protocol/nylon_ipc.proto
index 8cad7484..45ba8551 100644
--- a/protocol/nylon_ipc.proto
+++ b/protocol/nylon_ipc.proto
@@ -144,9 +144,9 @@ enum EndpointProbeStatus {
 
 message EndpointProbeResult {
   string address = 1;
-  optional string resolved = 2;
-  EndpointProbeStatus status = 3;
-  int64 latency_ns = 4;
+  optional string resolved = 4;
+  EndpointProbeStatus status = 5;
+  int64 latency_ns = 6;
 }
 
 message ProbeResponse {