diff --git a/core/router_harness.go b/core/router_harness.go index 23c32cf7..ceeb5174 100644 --- a/core/router_harness.go +++ b/core/router_harness.go @@ -34,7 +34,7 @@ func (m MockEndpoint) Node() state.NodeId { return m.node } -func (m MockEndpoint) UpdatePing(ping time.Duration) { +func (m *MockEndpoint) UpdatePing(ping time.Duration) { m.metric = min(uint32(ping.Microseconds()), state.INF) } diff --git a/e2e/harness.go b/e2e/harness.go index 8a7c5bce..0cb6fcdd 100644 --- a/e2e/harness.go +++ b/e2e/harness.go @@ -5,16 +5,16 @@ package e2e import ( "bytes" "context" + "errors" "fmt" "io" "net/netip" "os" "path/filepath" - "reflect" + "strings" "sync" "testing" "time" - "unsafe" "github.com/encodeous/nylon/protocol" "github.com/encodeous/nylon/state" @@ -47,6 +47,7 @@ type Harness struct { RootDir string Subnet string Gateway string + testDir string } // NewHarness creates a test harness with a unique subnet @@ -70,32 +71,12 @@ func NewHarness(t *testing.T) *Harness { rootDir = parent } - networkMu.Lock() - defer networkMu.Unlock() - - subnet, gateway, err := AllocateDockerSubnet(ctx) + newNetwork, subnet, gateway, err := createNetwork(ctx) if err != nil { t.Fatal(err) } t.Logf("Allocated subnet: %s, gateway: %s", subnet, gateway) - // Create network with a Docker-checked non-overlapping explicit subnet. - newNetwork, err := tcnetwork.New(ctx, - tcnetwork.WithAttachable(), - tcnetwork.WithDriver("bridge"), - tcnetwork.WithIPAM(&network.IPAM{ - Driver: "default", - Config: []network.IPAMConfig{ - { - Subnet: netip.MustParsePrefix(subnet), - Gateway: netip.MustParseAddr(gateway), - }, - }, - })) - if err != nil { - t.Fatal(err) - } - h := &Harness{ t: t, ctx: ctx, @@ -113,6 +94,45 @@ func NewHarness(t *testing.T) *Harness { return h } +// createNetwork creates a bridge network on a free subnet. networkMu only +// serializes allocation within this process, so another process (another test +// package or CI job on the same daemon) can still claim the subnet between +// listing networks and creating ours. Docker rejects that with an overlap +// error, so we allocate again and retry. +func createNetwork(ctx context.Context) (*testcontainers.DockerNetwork, string, string, error) { + networkMu.Lock() + defer networkMu.Unlock() + + var lastErr error + for range 5 { + subnet, gateway, err := AllocateDockerSubnet(ctx) + if err != nil { + return nil, "", "", err + } + // Create network with a Docker-checked non-overlapping explicit subnet. + nw, err := tcnetwork.New(ctx, + tcnetwork.WithAttachable(), + tcnetwork.WithDriver("bridge"), + tcnetwork.WithIPAM(&network.IPAM{ + Driver: "default", + Config: []network.IPAMConfig{ + { + Subnet: netip.MustParsePrefix(subnet), + Gateway: netip.MustParseAddr(gateway), + }, + }, + })) + if err == nil { + return nw, subnet, gateway, nil + } + if !strings.Contains(err.Error(), "overlaps") { + return nil, "", "", err + } + lastErr = err + } + return nil, "", "", fmt.Errorf("could not allocate a free subnet: %w", lastErr) +} + type NodeSpec struct { Name string IP string @@ -122,17 +142,37 @@ type NodeSpec struct { } func (h *Harness) StartNodes(specs ...NodeSpec) { + h.t.Helper() + // t.Fatal must only be called from the test goroutine, so the workers + // collect errors and we fail here once all of them are done. + errs := make([]error, len(specs)) var wg sync.WaitGroup - wg.Add(len(specs)) - for _, spec := range specs { - go func(s NodeSpec) { - defer wg.Done() - h.StartNode(s.Name, s.IP, s.CentralConfigPath, s.NodeConfigPath, s.ExtraArgs...) - }(spec) + for i, spec := range specs { + wg.Go(func() { + _, errs[i] = h.startNode(spec.Name, spec.IP, spec.CentralConfigPath, spec.NodeConfigPath, spec.ExtraArgs...) + }) } wg.Wait() + if err := errors.Join(errs...); err != nil { + h.t.Fatal(err) + } } func (h *Harness) StartNode(name string, ip string, centralConfigPath, nodeConfigPath string, extraArgs ...string) testcontainers.Container { + h.t.Helper() + cont, err := h.startNode(name, ip, centralConfigPath, nodeConfigPath, extraArgs...) + if err != nil { + h.t.Fatal(err) + } + return cont +} + +// containerName builds a Docker-safe container name. Subtest names contain +// '/', which Docker does not allow in names. +func (h *Harness) containerName(name string) string { + return strings.ReplaceAll(h.t.Name(), "/", "_") + "-" + name +} + +func (h *Harness) startNode(name string, ip string, centralConfigPath, nodeConfigPath string, extraArgs ...string) (testcontainers.Container, error) { h.t.Logf("Starting node %s at %s", name, ip) req := testcontainers.ContainerRequest{ Image: ImageName, @@ -175,19 +215,23 @@ func (h *Harness) StartNode(name string, ip string, centralConfigPath, nodeConfi &UnifiedLogConsumer{Node: name, Manager: h.LogManager}, }, }, - Name: h.t.Name() + "-" + name, + Name: h.containerName(name), } cont, err := testcontainers.GenericContainer(h.ctx, testcontainers.GenericContainerRequest{ ContainerRequest: req, Started: true, }) + // GenericContainer can return a container alongside an error (e.g. when + // the wait strategy times out), so track it for cleanup either way. + if cont != nil { + h.mu.Lock() + h.Nodes[name] = cont + h.mu.Unlock() + } if err != nil { - h.t.Fatalf("failed to start container %s: %v", name, err) + return cont, fmt.Errorf("failed to start container %s: %w", name, err) } - h.mu.Lock() - h.Nodes[name] = cont - h.mu.Unlock() - return cont + return cont, nil } func (h *Harness) WaitForLog(nodeName string, pattern string) { @@ -281,22 +325,42 @@ func (h *Harness) waitFor(nodeName string, source LogSource, pattern string, isR } } +// managerWriter feeds an exec stream into the LogManager one line at a time. +// The stream arrives in arbitrary chunks, so a partial line is held until its +// newline arrives; otherwise a pattern split across two chunks never matches. type managerWriter struct { node string source LogSource manager *LogManager + partial []byte } func (w *managerWriter) Write(p []byte) (n int, err error) { - content := StripAnsi(string(p)) - ts := time.Since(w.manager.start).Truncate(time.Millisecond) - fmt.Printf("[+%s][%s:%s] %s", ts, w.node, w.source, content) - w.manager.Accept(w.node, w.source, content) + w.partial = append(w.partial, p...) + for { + idx := bytes.IndexByte(w.partial, '\n') + if idx < 0 { + break + } + w.emit(string(w.partial[:idx+1])) + w.partial = w.partial[idx+1:] + } return len(p), nil } -func GetUnexportedField(field reflect.Value) interface{} { - return reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem().Interface() +// Flush emits any trailing output that did not end in a newline. +func (w *managerWriter) Flush() { + if len(w.partial) > 0 { + w.emit(string(w.partial) + "\n") + w.partial = nil + } +} + +func (w *managerWriter) emit(line string) { + content := StripAnsi(line) + ts := time.Since(w.manager.start).Truncate(time.Millisecond) + fmt.Printf("[+%s][%s:%s] %s", ts, w.node, w.source, content) + w.manager.Accept(w.node, w.source, content) } func (h *Harness) StartTrace(nodeName string) { @@ -319,25 +383,33 @@ func (h *Harness) StartTrace(nodeName string) { TTY: false, } - docker := cont.(*testcontainers.DockerContainer) - // this is very sketchy, but testcontainers doesn't provide any API - v := reflect.ValueOf(docker) - y := GetUnexportedField(v.Elem().FieldByName("provider")).(*testcontainers.DockerProvider) - cli := y.Client() + // container.Exec only returns output once the command exits, so the + // streaming exec goes through our own client. NewDockerClientWithOpts + // resolves the same daemon that testcontainers uses. + cli, err := testcontainers.NewDockerClientWithOpts(h.ctx) + if err != nil { + // t.Fatal is not allowed outside the test goroutine. + h.t.Errorf("Failed to start trace for %s: %v", nodeName, err) + return + } + defer cli.Close() execIDResp, err := cli.ExecCreate(h.ctx, cont.GetContainerID(), execOptions) if err != nil { - h.t.Fatalf("Failed to start trace for %s: %v", nodeName, err) + h.t.Errorf("Failed to start trace for %s: %v", nodeName, err) + return } resp, err := cli.ExecAttach(h.ctx, execIDResp.ID, client.ExecAttachOptions{}) if err != nil { - h.t.Fatalf("Failed to start trace for %s: %v", nodeName, err) + h.t.Errorf("Failed to start trace for %s: %v", nodeName, err) + return } defer resp.Close() stdoutWriter := &managerWriter{node: nodeName, source: SourceTrace, manager: h.LogManager} _, _ = stdcopy.StdCopy(stdoutWriter, stdoutWriter, resp.Reader) + stdoutWriter.Flush() }() } func (h *Harness) Cleanup() { @@ -500,7 +572,7 @@ func (h *Harness) StartDNS(name string, ip string, corefile string, zones map[st } } }, - Name: h.t.Name() + "-" + name, + Name: h.containerName(name), } container, err := testcontainers.GenericContainer(h.ctx, testcontainers.GenericContainerRequest{ @@ -518,14 +590,23 @@ func (h *Harness) StartDNS(name string, ip string, corefile string, zones map[st return container } -// SetupTestDir creates a directory for the current test run +// SetupTestDir returns the directory for the current test run. The first call +// clears out files left by a previous run; later calls return the same +// directory without wiping it, so helpers such as StartDNS can share it with +// the test. func (h *Harness) SetupTestDir() string { + h.mu.Lock() + defer h.mu.Unlock() + if h.testDir != "" { + return h.testDir + } dir := filepath.Join(h.RootDir, "e2e", "runs", h.t.Name()) // Clean up previous run os.RemoveAll(dir) if err := os.MkdirAll(dir, 0755); err != nil { h.t.Fatal(err) } + h.testDir = dir return dir } diff --git a/go.mod b/go.mod index 4cb87982..8e9e64b0 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/digineo/go-ping v1.2.0 github.com/dustin/go-broadcast v0.0.0-20211018055107-71439988bd91 github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd - github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4 + github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101215344-e479629b9d9a github.com/encodeous/tint v1.2.0 github.com/gaissmai/bart v0.25.0 github.com/goccy/go-yaml v1.19.1 diff --git a/go.sum b/go.sum index 77d6c7d8..ea771222 100644 --- a/go.sum +++ b/go.sum @@ -43,8 +43,8 @@ github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd h1:B32Ob80QTv5MomcVt709TsiWyD0QrpUYtnwW1jQFNlE= github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd/go.mod h1:DiXCPJtfZYioejF9zv9wfs3TXqWWglKGQ20DsBNVWVw= -github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4 h1:kr+P9WO7u+hPmOkx7XtuSA7Al8Ngdfr9KJc5gNQSDVo= -github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4/go.mod h1:SulefP9J0mccyOLDA+5Aa5nO569HF9YgkSRmx48wkQA= +github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101215344-e479629b9d9a h1:WCA0EK82cM8YKA8yVkxQ7KrNHVPweFI+QSeLeOLSIn4= +github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101215344-e479629b9d9a/go.mod h1:SulefP9J0mccyOLDA+5Aa5nO569HF9YgkSRmx48wkQA= github.com/encodeous/tint v1.2.0 h1:1Y+32Iu+C8MXBoNjsM4YDf6iAkcks7csAI9f7b4fr8k= github.com/encodeous/tint v1.2.0/go.mod h1:pyfyH+fKmtmIuWVWeMcSJjJKVJSb9sVlIfVsW3jkP+Q= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= diff --git a/integration/ack_retract_test.go b/integration/ack_retract_test.go index e51bbb16..fae16ee7 100644 --- a/integration/ack_retract_test.go +++ b/integration/ack_retract_test.go @@ -5,10 +5,11 @@ package integration import ( "net/netip" "testing" + "testing/synctest" "time" + "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/state" - "go.uber.org/goleak" ) // TestAckRetractSentOnWire verifies that ACK retract messages are actually @@ -20,110 +21,47 @@ import ( // If ACKs are sent correctly, B's RetractedBy will be populated and the // held route (blackhole) released quickly. func TestAckRetractSentOnWire(t *testing.T) { - defer goleak.VerifyNone(t) - tunables := state.DefaultRouterTunables() - tunables.ProbeDelay /= 3 - tunables.RouteUpdateDelay = 500 * time.Millisecond - tunables.NeighbourIOFlushDelay = 100 * time.Millisecond + synctest.Test(t, func(t *testing.T) { + tunables := state.DefaultRouterTunables() + tunables.ProbeDelay /= 3 + tunables.RouteUpdateDelay = 500 * time.Millisecond + tunables.NeighbourIOFlushDelay = 100 * time.Millisecond - vh := &VirtualHarness{} - vh.Tunables = &tunables - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") + h := NewHarness(t) + h.Tunables = &tunables + a1, b1 := "192.168.1.1:1234", "192.168.1.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + h.Start() - vh.Central.Graph = []string{ - "a, b", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - - errs := vh.Start() - - bPrefix := netip.MustParsePrefix("10.0.0.2/32") - na := vh.Nylons[0].Load() - nb := vh.Nylons[1].Load() - - // Wait for routes to converge: a should have a route to b's prefix - converged := make(chan struct{}) - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(50 * time.Millisecond): - na.Dispatch(func() error { - route, ok := na.RouterState.Routes[bPrefix] - if ok && route.Metric != state.INF { - select { - case <-converged: - default: - close(converged) - } - } - return nil - }) - } - } - }() + bPrefix := netip.MustParsePrefix("10.0.0.2/32") + h.Eventually(30*time.Second, "a to learn a route to b", func() bool { + return OnNode(h, "a", func(n *core.Nylon) bool { + route, ok := n.RouterState.Routes[bPrefix] + return ok && route.Metric != state.INF + }) + }) - select { - case <-converged: - t.Log("Routes converged") - case <-time.After(30 * time.Second): - t.Fatal("Timed out waiting for route convergence") - case err := <-errs: - t.Fatal(err) - } + // B retracts its prefix: remove from Advertised and broadcast INF. + OnNode(h, "b", func(n *core.Nylon) struct{} { + delete(n.RouterState.Advertised, bPrefix) + n.BroadcastSendRouteUpdate(state.PubRoute{ + Source: state.Source{NodeId: "b", Prefix: bPrefix}, + FD: state.FD{Seqno: n.RouterState.GetSeqno(bPrefix), Metric: state.INF}, + }) + return struct{}{} + }) - // B retracts its prefix: remove from Advertised and broadcast INF. - nb.Dispatch(func() error { - delete(nb.RouterState.Advertised, bPrefix) - nb.BroadcastSendRouteUpdate(state.PubRoute{ - Source: state.Source{NodeId: "b", Prefix: bPrefix}, - FD: state.FD{Seqno: nb.RouterState.GetSeqno(bPrefix), Metric: state.INF}, + // If ACKs are sent on wire, B will receive HandleAckRetract from A, + // populating RetractedBy. Since A is B's only neighbour, the held route + // will be released. + h.Eventually(5*time.Second, "b to receive an ACK retract on the wire", func() bool { + return OnNode(h, "b", func(n *core.Nylon) bool { + route, ok := n.RouterState.Routes[bPrefix] + return !ok || len(route.RetractedBy) > 0 + }) }) - return nil }) - - // If ACKs are sent on wire, B will receive HandleAckRetract from A, - // populating RetractedBy. Since A is B's only neighbour, the held route - // will be released. - ackReceived := make(chan struct{}) - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(50 * time.Millisecond): - nb.Dispatch(func() error { - route, ok := nb.RouterState.Routes[bPrefix] - if !ok || len(route.RetractedBy) > 0 { - select { - case <-ackReceived: - default: - close(ackReceived) - } - } - return nil - }) - } - } - }() - - select { - case <-ackReceived: - t.Log("ACK retract received on wire - held route released") - case <-time.After(5 * time.Second): - t.Fatal("ACK retract was never received on wire - bug: flushIO does not send ACKs") - case err := <-errs: - t.Fatal(err) - } - - vh.Stop() } diff --git a/integration/apply_config_test.go b/integration/apply_config_test.go index 4081e953..addd4ac8 100644 --- a/integration/apply_config_test.go +++ b/integration/apply_config_test.go @@ -6,332 +6,203 @@ import ( "fmt" "net/netip" "testing" + "testing/synctest" "time" "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/state" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "go.uber.org/goleak" ) func TestApplyCentralConfigRemovesNeighbourFromLiveNode(t *testing.T) { - defer goleak.VerifyNone(t) - - vh := &VirtualHarness{} - a1 := "192.168.10.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.10.2:1234" - vh.NewNode("b", "10.0.0.2/32") - c1 := "192.168.10.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(a1, c1) - vh.AddLink(c1, a1) - vh.AddLink(b1, c1) - vh.AddLink(c1, b1) - - errs := vh.Start() - defer vh.Stop() - - var apply applyResult - done := make(chan struct{}) - next := vh.Central - next.Timestamp++ - next.Graph = []string{ - "a, c", - "b, c", - } - - a := vh.Nylons[vh.IndexOf("a")].Load() - a.Dispatch(func() error { - beforeC := a.RouterState.GetNeighbour("c") - if beforeC == nil || len(beforeC.Eps) == 0 { - apply.Err = fmt.Errorf("expected neighbour c with at least one endpoint before apply") - close(done) - return nil + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + a1, b1, c1 := "192.168.10.1:1234", "192.168.10.2:1234", "192.168.10.3:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.AddNode("c", "10.0.0.3/32", c1) + h.Central.Graph = []string{"a, b, c"} + h.Connect(a1, b1) + h.Connect(a1, c1) + h.Connect(b1, c1) + h.Start() + + next := h.Central + next.Timestamp++ + next.Graph = []string{ + "a, c", + "b, c", } - keptEndpoint := beforeC.Eps[0] - result, err := a.ApplyCentralConfig(&next) - afterC := a.RouterState.GetNeighbour("c") + bKey := h.Router("b").PubKey + cKey := h.Router("c").PubKey - apply = applyResult{ - Result: result, - Err: err, - HasBNeighbour: a.RouterState.GetNeighbour("b") != nil, - HasCNeighbour: afterC != nil, - HasBAppliedPeer: hasAppliedPeer(a, "b"), - HasCAppliedPeer: hasAppliedPeer(a, "c"), - HasBWGPeer: hasWGPeer(a, vh.Central.GetNode("b").PubKey), - HasCWGPeer: hasWGPeer(a, vh.Central.GetNode("c").PubKey), - PreservedEndpoint: afterC != nil && len(afterC.Eps) > 0 && afterC.Eps[0] == keptEndpoint, - } - close(done) - return nil - }) - - select { - case <-done: - case err := <-errs: - t.Fatal(err) - case <-time.After(5 * time.Second): - t.Fatal("timed out waiting for central config apply") - } + apply := OnNode(h, "a", func(n *core.Nylon) applyResult { + beforeC := n.RouterState.GetNeighbour("c") + if beforeC == nil || len(beforeC.Eps) == 0 { + return applyResult{Err: fmt.Errorf("expected neighbour c with at least one endpoint before apply")} + } + keptEndpoint := beforeC.Eps[0] + result, err := n.ApplyCentralConfig(&next) + afterC := n.RouterState.GetNeighbour("c") + return applyResult{ + Result: result, + Err: err, + HasBNeighbour: n.RouterState.GetNeighbour("b") != nil, + HasCNeighbour: afterC != nil, + HasBAppliedPeer: hasAppliedPeer(n, "b"), + HasCAppliedPeer: hasAppliedPeer(n, "c"), + HasBWGPeer: hasWGPeer(n, bKey), + HasCWGPeer: hasWGPeer(n, cKey), + PreservedEndpoint: afterC != nil && len(afterC.Eps) > 0 && afterC.Eps[0] == keptEndpoint, + } + }) - assert.NoError(t, apply.Err) - assert.Equal(t, core.ApplyApplied, apply.Result) - assert.False(t, apply.HasBNeighbour) - assert.True(t, apply.HasCNeighbour) - assert.False(t, apply.HasBAppliedPeer) - assert.True(t, apply.HasCAppliedPeer) - assert.False(t, apply.HasBWGPeer) - assert.True(t, apply.HasCWGPeer) - assert.True(t, apply.PreservedEndpoint) + assert.NoError(t, apply.Err) + assert.Equal(t, core.ApplyApplied, apply.Result) + assert.False(t, apply.HasBNeighbour) + assert.True(t, apply.HasCNeighbour) + assert.False(t, apply.HasBAppliedPeer) + assert.True(t, apply.HasCAppliedPeer) + assert.False(t, apply.HasBWGPeer) + assert.True(t, apply.HasCWGPeer) + assert.True(t, apply.PreservedEndpoint) + }) } func TestApplyCentralConfigLocalNodeRemovedRequiresRestart(t *testing.T) { - defer goleak.VerifyNone(t) - - vh := &VirtualHarness{} - a1 := "192.168.20.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.20.2:1234" - 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) - vh.AddLink(b1, a1) - - errs := vh.Start() - defer vh.Stop() - - a := vh.Nylons[vh.IndexOf("a")].Load() - next := vh.Central - next.Timestamp++ - next.Routers = next.Routers[1:] - next.Graph = nil + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + a1, b1 := "192.168.20.1:1234", "192.168.20.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + h.Start() + + next := h.Central + next.Timestamp++ + next.Routers = next.Routers[1:] + next.Graph = nil + + type outcome struct { + result core.ApplyResult + err error + centralUnchanged bool + bStillNeighbour bool + } + got := OnNode(h, "a", func(n *core.Nylon) outcome { + result, err := n.ApplyCentralConfig(&next) + return outcome{ + result: result, + err: err, + centralUnchanged: n.CentralCfg.Timestamp == h.Central.Timestamp, + bStillNeighbour: n.RouterState.GetNeighbour("b") != nil, + } + }) - done := make(chan struct{}) - var result core.ApplyResult - var err error - var centralUnchanged bool - var bStillNeighbour bool - a.Dispatch(func() error { - result, err = a.ApplyCentralConfig(&next) - centralUnchanged = a.CentralCfg.Timestamp == vh.Central.Timestamp - bStillNeighbour = a.RouterState.GetNeighbour("b") != nil - close(done) - return nil + assert.Equal(t, core.ApplyRestartRequired, got.result) + assert.Error(t, got.err) + assert.True(t, got.centralUnchanged) + assert.True(t, got.bStillNeighbour) }) - - select { - case <-done: - case err := <-errs: - t.Fatal(err) - case <-time.After(5 * time.Second): - t.Fatal("timed out waiting for rejected central config apply") - } - - assert.Equal(t, core.ApplyRestartRequired, result) - assert.Error(t, err) - assert.True(t, centralUnchanged) - assert.True(t, bStillNeighbour) } func TestApplyCentralConfigRotatesPeerKeyWithoutChangingNextHop(t *testing.T) { - defer goleak.VerifyNone(t) - - vh := &VirtualHarness{UntrackedRouting: true} - a1 := "192.168.30.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.30.2:1234" - 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) - vh.AddLink(b1, a1) - - errs := vh.Start() - defer vh.Stop() - - received := make(chan byte, 16) - vh.Net.SelfHandler = func(node state.NodeId, _, dst netip.Addr, data []byte) bool { - if node == "b" && dst == netip.MustParseAddr("10.0.0.2") && len(data) != 0 { - select { - case received <- data[0]: - default: + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.UntrackedRouting = true + a1, b1 := "192.168.30.1:1234", "192.168.30.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + + received := make(chan byte, 16) + bAddr := netip.MustParseAddr("10.0.0.2") + h.OnDeliver(func(node state.NodeId, _, dst netip.Addr, data []byte) bool { + if node == "b" && dst == bAddr && len(data) != 0 { + select { + case received <- data[0]: + default: + } } - } - return true - } - - waitForPayload(t, vh, errs, received, 1) + return true + }) + h.Start() - newPrivateKey := state.GenerateKey() - bIdx := vh.IndexOf("b") - b := vh.Nylons[bIdx].Load() - err, next := vh.Central.Clone() - require.NoError(t, err) - next.Timestamp++ - next.Routers[bIdx].PubKey = newPrivateKey.Pubkey() + waitForPayload(h, received, 1) - // A local identity is fixed at construction. Restart the remote node while - // keeping the sender alive so its forwarding table must rebind the same hop. - b.Stop() - select { - case <-vh.nodeDone[bIdx]: - case <-time.After(5 * time.Second): - t.Fatal("timed out stopping remote node") - } - require.NoError(t, vh.Net.binds[bIdx].Close()) - local := vh.Local[bIdx] - local.Key = newPrivateKey - // Packet delivery holds linksMu while accessing bind slots. Pause delivery - // while the restarted node installs its replacement virtual bind and TUN. - vh.linksMu.Lock() - replacement, err := core.NewNylon(*next, local, *vh.LogLevel, "", map[string]any{"vnet": vh.Net}, state.NylonOptions{DBG_log_wireguard: true}, vh.Tunables) - vh.linksMu.Unlock() - require.NoError(t, err) - vh.Nylons[bIdx].Store(replacement) - done := make(chan struct{}) - vh.nodeDone[bIdx] = done - go func() { - defer close(done) - if err := replacement.Start(); err != nil { - errs <- err - } - }() + newPrivateKey := state.GenerateKey() + err, next := h.Central.Clone() + require.NoError(t, err) + next.Timestamp++ + routerOf(next, "b").PubKey = newPrivateKey.Pubkey() - a := vh.Nylons[vh.IndexOf("a")].Load() - applyConfigAndWait(t, a, next) + // A local identity is fixed at construction. Restart the remote node + // while keeping the sender alive so its forwarding table must rebind + // the same hop. + h.RestartNode("b", *next, func(local *state.LocalCfg) { local.Key = newPrivateKey }) + h.ApplyConfig("a", next) - waitForPayload(t, vh, errs, received, 2) + waitForPayload(h, received, 2) + }) } func TestApplyCentralConfigRemovedPrefixStopsSeqnoRequests(t *testing.T) { - defer goleak.VerifyNone(t) - - vh := &VirtualHarness{} - a1 := "192.168.40.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.40.2:1234" - 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) - vh.AddLink(b1, a1) - removed := netip.MustParsePrefix("10.1.0.2/32") - bi := vh.IndexOf("b") - vh.Central.Routers[bi].Prefixes = append(vh.Central.Routers[bi].Prefixes, state.PrefixHealthWrapper{ - PrefixHealth: &state.StaticPrefixHealth{Prefix: removed}, - }) - - errs := vh.Start() - defer vh.Stop() - - a := vh.Nylons[vh.IndexOf("a")].Load() - b := vh.Nylons[bi].Load() - neighHasRoute := func() bool { - res := make(chan bool, 1) - a.Dispatch(func() error { - route, ok := a.RouterState.GetNeighbour("b").Routes[removed] - res <- ok && route.Metric != state.INF - return nil + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + a1, b1 := "192.168.40.1:1234", "192.168.40.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + removed := netip.MustParsePrefix("10.1.0.2/32") + h.Router("b").Prefixes = append(h.Router("b").Prefixes, state.PrefixHealthWrapper{ + PrefixHealth: &state.StaticPrefixHealth{Prefix: removed}, }) - select { - case ok := <-res: - return ok - case err := <-errs: - t.Fatal(err) - case <-time.After(5 * time.Second): - t.Fatal("timed out waiting for dispatch") - } - return false - } - deadline := time.Now().Add(10 * time.Second) - for !neighHasRoute() { - if time.Now().After(deadline) { - t.Fatal("timed out waiting for a to learn the prefix") - } - time.Sleep(100 * time.Millisecond) - } - - err, next := vh.Central.Clone() - assert.NoError(t, err) - next.Timestamp++ - next.Routers[bi].Prefixes = next.Routers[bi].Prefixes[:1] - applyConfigAndWait(t, b, next) - applyConfigAndWait(t, a, next) + h.Start() - // stale route keeps a requesting seqnos until it expires - done := make(chan struct{}) - var stale bool - a.Dispatch(func() error { - _, stale = a.RouterState.GetNeighbour("b").Routes[removed] - close(done) - return nil - }) - <-done - assert.False(t, stale) -} + h.Eventually(10*time.Second, "a to learn the prefix", func() bool { + return OnNode(h, "a", func(n *core.Nylon) bool { + route, ok := n.RouterState.GetNeighbour("b").Routes[removed] + return ok && route.Metric != state.INF + }) + }) -func applyConfigAndWait(t *testing.T, n *core.Nylon, cfg *state.CentralCfg) { - t.Helper() - done := make(chan struct{}) - var result core.ApplyResult - var err error - n.Dispatch(func() error { - result, err = n.ApplyCentralConfig(cfg) - close(done) - return nil + err, next := h.Central.Clone() + require.NoError(t, err) + next.Timestamp++ + bNext := routerOf(next, "b") + bNext.Prefixes = bNext.Prefixes[:1] + h.ApplyConfig("b", next) + h.ApplyConfig("a", next) + + // a stale route would keep a requesting seqnos until it expires + stale := OnNode(h, "a", func(n *core.Nylon) bool { + _, ok := n.RouterState.GetNeighbour("b").Routes[removed] + return ok + }) + assert.False(t, stale) }) - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("timed out waiting for central config apply") - } - assert.NoError(t, err) - assert.Equal(t, core.ApplyApplied, result) } -func waitForPayload(t *testing.T, vh *VirtualHarness, errs <-chan error, received <-chan byte, payload byte) { - t.Helper() - timeout := time.NewTimer(10 * time.Second) - defer timeout.Stop() - ticker := time.NewTicker(100 * time.Millisecond) - defer ticker.Stop() +// waitForPayload keeps sending payload from a to b until b receives it. +func waitForPayload(h *Harness, received <-chan byte, payload byte) { + h.t.Helper() + stop := h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.2", []byte{payload}, 64) + defer stop() + deadline := time.After(10 * time.Second) for { select { case got := <-received: if got == payload { return } - case <-ticker.C: - vh.Net.Send("a", "10.0.0.1", "10.0.0.2", []byte{payload}, 64) - case err := <-errs: - t.Fatal(err) - case <-timeout.C: - t.Fatalf("timed out waiting for payload %d", payload) + case <-deadline: + h.t.Fatalf("timed out waiting for payload %d", payload) + case <-h.ctx.Done(): + h.t.Fatalf("harness stopped while waiting for payload %d", payload) } } } @@ -363,3 +234,13 @@ func hasWGPeer(n *core.Nylon, key state.NyPublicKey) bool { } return false } + +// routerOf returns id's entry in cfg. +func routerOf(cfg *state.CentralCfg, id state.NodeId) *state.RouterCfg { + for i := range cfg.Routers { + if cfg.Routers[i].Id == id { + return &cfg.Routers[i] + } + } + panic("unknown router " + string(id)) +} diff --git a/integration/basic_test.go b/integration/basic_test.go index 9c6e448c..478e13f0 100644 --- a/integration/basic_test.go +++ b/integration/basic_test.go @@ -5,139 +5,74 @@ package integration import ( "net/netip" "testing" + "testing/synctest" "time" "github.com/encodeous/nylon/state" - "go.uber.org/goleak" ) func TestStartStop(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - vh.NewNode("node1", "10.0.0.1/32") - vh.NewNode("node2", "10.0.0.2/32") - vh.NewNode("node3", "10.0.0.3/32") - vh.Central.Graph = []string{ - "node1, node2, node3", - } - errs := vh.Start() - select { - case <-time.After(1000 * time.Millisecond): - case err := <-errs: - t.Error(err) - } - vh.Stop() + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.AddNode("node1", "10.0.0.1/32") + h.AddNode("node2", "10.0.0.2/32") + h.AddNode("node3", "10.0.0.3/32") + h.Central.Graph = []string{"node1, node2, node3"} + h.Start() + time.Sleep(time.Second) + }) } func TestSimplePing(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - 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) - vh.AddLink(b1, a1) - - errs := vh.Start() - - vn := vh.Net - cc := make(chan bool, 100) - - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "b" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.2" && data[0] == 111 { - cc <- true - } - return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vn.Send("a", "10.0.0.1", "10.0.0.2", []byte{111}, 64) - } - } - }() - - select { - case <-cc: - t.Log("Got ping!") - case <-time.After(10 * time.Second): - t.Error("Timed out waiting for ping") - case err := <-errs: - t.Error(err) - } - vh.Stop() + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + a1, b1 := "192.168.1.1:1234", "192.168.1.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + + got := expectPacket(h, "b", "10.0.0.1", "10.0.0.2", 111) + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.2", []byte{111}, 64) + await(h, got, 10*time.Second, "ping from a to b") + }) } func TestSimpleRoutedPing(t *testing.T) { - defer goleak.VerifyNone(t) - tunables := state.DefaultRouterTunables() - tunables.ProbeDelay /= 3 // 3x faster - tunables.RouteUpdateDelay /= 3 - - vh := &VirtualHarness{} - vh.Tunables = &tunables - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") - c1 := "192.168.1.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b", - "b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(b1, c1) - vh.AddLink(c1, b1) - - errs := vh.Start() - - vn := vh.Net - cc := make(chan bool, 100) + synctest.Test(t, func(t *testing.T) { + tunables := state.DefaultRouterTunables() + tunables.ProbeDelay /= 3 // 3x faster + tunables.RouteUpdateDelay /= 3 + + h := NewHarness(t) + h.Tunables = &tunables + a1, b1, c1 := "192.168.1.1:1234", "192.168.1.2:1234", "192.168.1.3:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.AddNode("c", "10.0.0.3/32", c1) + h.Central.Graph = []string{"a, b", "b, c"} + h.Connect(a1, b1) + h.Connect(b1, c1) + + got := expectPacket(h, "c", "10.0.0.1", "10.0.0.3", 222) + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) + await(h, got, 100*time.Second, "ping from a to c through b") + }) +} - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "c" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 { - cc <- true +// expectPacket returns a channel that receives once node has been delivered a +// packet from src to dst whose payload starts with marker. It replaces any +// OnDeliver filter. +func expectPacket(h *Harness, node state.NodeId, src, dst string, marker byte) <-chan struct{} { + got := NewSignal() + srcAddr, dstAddr := netip.MustParseAddr(src), netip.MustParseAddr(dst) + h.OnDeliver(func(at state.NodeId, s, d netip.Addr, data []byte) bool { + if at == node && s == srcAddr && d == dstAddr && len(data) > 0 && data[0] == marker { + got.Trigger() } return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vn.Send("a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) - } - } - }() - - select { - case <-cc: - t.Log("Got ping!") - case <-time.After(100 * time.Second): - t.Error("Timed out waiting for ping") - case err := <-errs: - t.Error(err) - } - vh.Stop() + }) + return got } diff --git a/integration/convergence_test.go b/integration/convergence_test.go index 8b1c193f..c8cb3948 100644 --- a/integration/convergence_test.go +++ b/integration/convergence_test.go @@ -3,100 +3,60 @@ package integration import ( - "fmt" "net/netip" "testing" + "testing/synctest" "time" "github.com/encodeous/nylon/state" - "go.uber.org/goleak" ) +// TestOptimalConvergence starts with a <-10-> b and a <-50-> c, so a reaches c +// directly. Once that works, it adds b <-10-> c and expects a to switch to the +// cheaper path through b. func TestOptimalConvergence(t *testing.T) { - defer goleak.VerifyNone(t) - tunables := state.DefaultRouterTunables() - tunables.ProbeDelay /= 3 // 3x faster - tunables.RouteUpdateDelay /= 3 - tunables.MinimumConfidenceWindow /= 5 - - vh := &VirtualHarness{} - vh.Tunables = &tunables - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") - c1 := "192.168.1.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - // a <-10-> b - vh.AddLink(a1, b1).WithLatency(10*time.Millisecond, 0) - vh.AddLink(b1, a1).WithLatency(10*time.Millisecond, 0) - - // c <-50-> a - vh.AddLink(a1, c1).WithLatency(50*time.Millisecond, 0) - vh.AddLink(c1, a1).WithLatency(50*time.Millisecond, 0) - - errs := vh.Start() - - vn := vh.Net - - conv1 := NewSignal() // first stage convergence: c <-50-> a <-10-> b - conv2 := NewSignal() // second stage convergence: a <-10-> b <-10-> c - success := NewSignal() - - go func() { - conv1.Wait() - - fmt.Printf("---- Stage 1 convergence reached ----\n") - - // b <-10-> c - vh.AddLink(b1, c1).WithLatency(10*time.Millisecond, 0) - vh.AddLink(c1, b1).WithLatency(10*time.Millisecond, 0) - - }() - - vn.TransitHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "b" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 && conv1.Triggered() { - // this means the network has reached the optimal path - conv2.Trigger() - } - return false // don't intercept the packet - } - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "c" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 { - conv1.Trigger() - if conv2.Triggered() { - success.Trigger() + synctest.Test(t, func(t *testing.T) { + tunables := state.DefaultRouterTunables() + tunables.ProbeDelay /= 3 // 3x faster + tunables.RouteUpdateDelay /= 3 + tunables.MinimumConfidenceWindow /= 5 + + h := NewHarness(t) + h.Tunables = &tunables + a1, b1, c1 := "192.168.1.1:1234", "192.168.1.2:1234", "192.168.1.3:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.AddNode("c", "10.0.0.3/32", c1) + h.Central.Graph = []string{"a, b, c"} + h.Connect(a1, b1).WithLatency(10*time.Millisecond, 0) + h.Connect(a1, c1).WithLatency(50*time.Millisecond, 0) + + src, dst := netip.MustParseAddr("10.0.0.1"), netip.MustParseAddr("10.0.0.3") + isProbe := func(s, d netip.Addr, data []byte) bool { return s == src && d == dst && data[0] == 222 } + direct := NewSignal() // a reached c over the direct link + viaB := NewSignal() // after that, a packet went through b + optimal := NewSignal() // and reached c + h.OnTransit(func(node state.NodeId, s, d netip.Addr, data []byte) bool { + if node == "b" && isProbe(s, d, data) && direct.Triggered() { + viaB.Trigger() } - } - return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vn.Send("a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) + return false // route it on as usual + }) + h.OnDeliver(func(node state.NodeId, s, d netip.Addr, data []byte) bool { + if node == "c" && isProbe(s, d, data) { + direct.Trigger() + if viaB.Triggered() { + optimal.Trigger() + } } - } - }() + return true + }) + + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) - select { - case <-success: - t.Log("Reached optimal!") - case <-time.After(200 * time.Second): - t.Error("Timed out waiting for ping") - case err := <-errs: - t.Error(err) - } - vh.Stop() + await(h, direct, 200*time.Second, "a to reach c directly") + h.Connect(b1, c1).WithLatency(10*time.Millisecond, 0) + await(h, optimal, 200*time.Second, "a to route to c through b") + }) } diff --git a/integration/harness.go b/integration/harness.go index 977470e4..5a5d1ddf 100644 --- a/integration/harness.go +++ b/integration/harness.go @@ -1,5 +1,12 @@ //go:build integration +// Package integration runs several nylon nodes in one process over a simulated +// network. +// +// Tests run inside a testing/synctest bubble. Time in a bubble is virtual and +// only moves forward once every goroutine in it is blocked, so protocol timers, +// link latency and test timeouts take no wall-clock time, and a node that +// would hang forever is reported as a deadlock instead. package integration import ( @@ -11,10 +18,14 @@ import ( "math/rand/v2" "net" "net/netip" + "os" + "os/signal" "runtime/pprof" "slices" "sync" "sync/atomic" + "syscall" + "testing" "time" "github.com/encodeous/nylon/core" @@ -27,327 +38,543 @@ import ( "github.com/encodeous/polyamide-wireguard-go/tun/tuntest" ) -type Signal chan bool +func init() { + // Nylon.Start calls signal.Notify, and the first Notify in a process + // starts a goroutine that never exits. Started inside a bubble it would + // belong to that bubble and keep it from finishing, so start it here. + signal.Notify(make(chan os.Signal, 1), syscall.SIGINT, syscall.SIGTERM) +} + +const ( + // startStagger spaces out node startup, like nodes coming up one by one. + startStagger = 500 * time.Millisecond + // pollInterval is how often Eventually re-checks its condition. + pollInterval = 50 * time.Millisecond + // dispatchTimeout bounds how long OnNode waits for a node's dispatcher. + dispatchTimeout = 5 * time.Second + // maxTestTime is the virtual time a test may run for. Running nodes keep + // timers pending, so a test stuck on a wait with no timeout never looks + // idle to synctest; the watchdog stops the nodes after this long so + // synctest can report the deadlock. It is a few seconds of real time. + maxTestTime = 15 * time.Minute + ipv4HeaderLen = 20 +) + +// Harness owns the nodes and the simulated network of one test. +// +// Build the topology with AddNode and Connect (or Link), adjust Central, Local +// or Tunables if needed, then call Start. The harness stops itself when the +// test ends. +type Harness struct { + t *testing.T + + Central state.CentralCfg + Local []state.LocalCfg + Tunables *state.RouterTunables + LogLevel slog.Level + // UntrackedRouting turns off UseSystemRouting on every node, so packets + // for peers stay inside polyamide instead of coming out of the TUN. + UntrackedRouting bool + + ctx context.Context + cancel context.CancelCauseFunc + ready chan struct{} // closed once every node is constructed + nodes []*node + wg sync.WaitGroup // harness goroutines: senders, links, binds, TUNs + stop sync.Once + + // epOwner and firstEp are written by AddNode and only read after Start. + epOwner map[netip.AddrPort]state.NodeId + firstEp map[state.NodeId]netip.AddrPort -func NewSignal() Signal { - return make(chan bool) + mu sync.RWMutex // guards links, binds and tuns + links map[linkKey]*Link + binds []conn.Bind + tuns []*tuntest.ChannelTUN + + onDeliver atomic.Pointer[PacketFilter] + onTransit atomic.Pointer[PacketFilter] + + rngMu sync.Mutex + rng *rand.Rand } -func (s Signal) Trigger() { - select { - case <-s: - default: - close(s) + +type node struct { + nylon atomic.Pointer[core.Nylon] + done chan struct{} // closed when the node's Start returns +} + +// NewHarness creates an empty harness that stops itself when t ends. +func NewHarness(t *testing.T) *Harness { + seed := rand.Uint64() + t.Logf("harness seed: %d", seed) + h := &Harness{ + t: t, + LogLevel: slog.LevelDebug, + epOwner: make(map[netip.AddrPort]state.NodeId), + firstEp: make(map[state.NodeId]netip.AddrPort), + links: make(map[linkKey]*Link), + rng: rand.New(rand.NewPCG(seed, seed)), } + t.Cleanup(h.Stop) + return h } -func (s Signal) Triggered() bool { - select { - case <-s: - return true - default: - return false + +// AddNode adds a router that advertises prefix and is reachable at endpoints. +// Its first endpoint is also the source address of every packet it sends. +func (h *Harness) AddNode(id state.NodeId, prefix string, endpoints ...string) { + key := state.GenerateKey() + router := state.RouterCfg{ + NodeCfg: state.NodeCfg{ + Id: id, + PubKey: key.Pubkey(), + Prefixes: []state.PrefixHealthWrapper{ + {PrefixHealth: &state.StaticPrefixHealth{Prefix: netip.MustParsePrefix(prefix)}}, + }, + }, + Endpoints: endpoints, } + for i, ep := range endpoints { + addr := netip.MustParseAddrPort(ep) + h.epOwner[addr] = id + if i == 0 { + h.firstEp[id] = addr + } + } + h.Central.Routers = append(h.Central.Routers, router) + h.Local = append(h.Local, state.LocalCfg{ + Key: key, + Id: id, + Port: 25565, + NoNetConfigure: true, + }) } -func (s Signal) Wait() { - <-s + +func (h *Harness) indexOf(id state.NodeId) int { + idx := slices.IndexFunc(h.Central.Routers, func(cfg state.RouterCfg) bool { return cfg.Id == id }) + if idx < 0 { + h.t.Fatalf("unknown node %s", id) + } + return idx } -type VirtualLink struct { - Edge state.Pair[bindtest.ChannelEndpoint2, bindtest.ChannelEndpoint2] - Latency time.Duration - Jitter time.Duration - PacketLoss float64 +// Router returns id's entry in the central config, for changes before Start. +func (h *Harness) Router(id state.NodeId) *state.RouterCfg { + return &h.Central.Routers[h.indexOf(id)] } -func (v *VirtualLink) simulate(pkt []byte, len int, from, to bindtest.ChannelEndpoint2, i *InMemoryNetwork) { - //fmt.Printf("begin send: %s -> %s\n", from.DstToString(), to.DstToString()) - if rand.Float64() < v.PacketLoss { - // drop - //fmt.Printf("dropped send: %s -> %s\n", from.DstToString(), to.DstToString()) - return +// LocalCfg returns id's local config, for changes before Start. +func (h *Harness) LocalCfg(id state.NodeId) *state.LocalCfg { + return &h.Local[h.indexOf(id)] +} + +// Nylon returns the running instance of node id. +func (h *Harness) Nylon(id state.NodeId) *core.Nylon { + return h.nodes[h.indexOf(id)].nylon.Load() +} + +// Start constructs every node, then starts them one startStagger apart. +func (h *Harness) Start() { + h.t.Helper() + h.ctx, h.cancel = context.WithCancelCause(context.Background()) + h.ready = make(chan struct{}) + h.nodes = make([]*node, len(h.Central.Routers)) + h.binds = make([]conn.Bind, len(h.Central.Routers)) + h.tuns = make([]*tuntest.ChannelTUN, len(h.Central.Routers)) + for idx := range h.nodes { + h.nodes[idx] = &node{} + h.Local[idx].UseSystemRouting = !h.UntrackedRouting } - toIdx := i.cfg.IndexOf(i.cfg.Endpoints[to.DstToString()]) - if v.Latency != 0 { - simJitter := rand.Float64() * float64(v.Jitter.Nanoseconds()) - simLat := v.Latency + time.Duration(simJitter) - go func() { // this is quite costly, but should be fine as long as we don't send packets too fast - select { - case <-i.cfg.Context.Done(): - return - case <-time.After(simLat): - //fmt.Printf("delay sent: %s -> %s\n", from.DstToString(), to.DstToString()) - err := i.binds[toIdx].Send([][]byte{pkt[:len]}, from) - if err != nil && !errors.Is(err, net.ErrClosed) { - panic(err) - } - } - }() - } else { - //fmt.Printf("sent: %s -> %s\n", from.DstToString(), to.DstToString()) - err := i.binds[toIdx].Send([][]byte{pkt[:len]}, from) - if err != nil && !errors.Is(err, net.ErrClosed) { - panic(err) + for idx := range h.nodes { + n, err := h.newNylon(h.Central, h.Local[idx]) + if err != nil { + // Nodes built so far own binds and TUNs, so tear them down. + close(h.ready) + h.Stop() + h.t.Fatalf("construct node %s: %v", h.Local[idx].Id, err) } + h.nodes[idx].nylon.Store(n) } -} + close(h.ready) -func (v *VirtualLink) WithLatency(lat, jitter time.Duration) *VirtualLink { - v.Latency = lat - v.Jitter = jitter - return v + for idx, nd := range h.nodes { + h.launch(nd, nd.nylon.Load(), time.Duration(idx)*startStagger) + } + h.spawn(h.watchdog) } -func (v *VirtualLink) WithPacketLoss(loss float64) *VirtualLink { - v.PacketLoss = loss - return v +func (h *Harness) watchdog() { + select { + case <-h.ctx.Done(): + case <-time.After(maxTestTime): + h.fail("test still running after %s of virtual time, stopping nodes so synctest can report what is blocked", maxTestTime) + for _, nd := range h.nodes { + if n := nd.nylon.Load(); n != nil { + n.Stop() + } + } + } } -type VirtualHarness struct { - Central state.CentralCfg - Context context.Context - Cancel context.CancelCauseFunc - Local []state.LocalCfg - Net *InMemoryNetwork - Nylons []atomic.Pointer[core.Nylon] - nodeDone []chan struct{} - Links []*VirtualLink - linksMu sync.RWMutex - Endpoints map[string]state.NodeId - UntrackedRouting bool - LogLevel *slog.Level - Tunables *state.RouterTunables +func (h *Harness) newNylon(central state.CentralCfg, local state.LocalCfg) (*core.Nylon, error) { + return core.NewNylon(central, local, h.LogLevel, "", map[string]any{"vnet": h}, + state.NylonOptions{DBG_log_wireguard: true}, h.Tunables) } -func (v *VirtualHarness) IndexOf(id state.NodeId) int { - return slices.IndexFunc(v.Central.Routers, func(cfg state.RouterCfg) bool { - return cfg.Id == id - }) +// launch runs n.Start in the background after delay. +func (h *Harness) launch(nd *node, n *core.Nylon, delay time.Duration) { + nd.done = make(chan struct{}) + done := nd.done + go func() { + defer close(done) + timer := time.NewTimer(delay) + select { + case <-timer.C: + case <-h.ctx.Done(): + // Start anyway: it returns at once on a stopped node, and its + // exit path releases what NewNylon allocated. + timer.Stop() + } + pprof.Do(context.Background(), pprof.Labels("nylon node", string(n.LocalCfg.Id)), func(context.Context) { + if err := n.Start(); err != nil { + h.fail("node %s: %v", n.LocalCfg.Id, err) + } + }) + }() } -func (v *VirtualHarness) NewNode(id state.NodeId, virtPrefix string) { - privKey := state.GenerateKey() - locCfg := state.LocalCfg{ - Key: privKey, - Id: id, - Port: 25565, - NoNetConfigure: true, - UseSystemRouting: !v.UntrackedRouting, +// RestartNode stops node id and starts a fresh instance from central and the +// node's local config after mutate (if non-nil) has edited it. The rest of the +// network keeps running. +func (h *Harness) RestartNode(id state.NodeId, central state.CentralCfg, mutate func(*state.LocalCfg)) { + h.t.Helper() + idx := h.indexOf(id) + nd := h.nodes[idx] + nd.nylon.Load().Stop() + await(h, nd.done, dispatchTimeout, "node "+string(id)+" to stop") + + h.mu.Lock() + old := h.binds[idx] + h.binds[idx] = nil + h.mu.Unlock() + if err := old.Close(); err != nil { + h.t.Fatalf("close bind of %s: %v", id, err) } - ncfg := state.RouterCfg{ - NodeCfg: state.NodeCfg{ - Id: id, - PubKey: privKey.Pubkey(), - Prefixes: []state.PrefixHealthWrapper{ - { - &state.StaticPrefixHealth{ - Prefix: netip.MustParsePrefix(virtPrefix), - Metric: 0, - }, - }, - }, - }, + + if mutate != nil { + mutate(&h.Local[idx]) + } + n, err := h.newNylon(central, h.Local[idx]) + if err != nil { + h.t.Fatalf("reconstruct node %s: %v", id, err) } - v.Central.Routers = append(v.Central.Routers, ncfg) - v.Local = append(v.Local, locCfg) -} - -func (v *VirtualHarness) AddLink(from, to string) *VirtualLink { - link := &VirtualLink{} - link.Edge = state.Pair[bindtest.ChannelEndpoint2, bindtest.ChannelEndpoint2]{ - V1: bindtest.ChannelEndpoint2(netip.MustParseAddrPort(from)), - V2: bindtest.ChannelEndpoint2(netip.MustParseAddrPort(to)), - } - v.linksMu.Lock() - v.Links = append(v.Links, link) - v.linksMu.Unlock() - return link -} - -func (v *VirtualHarness) Start() chan error { - ctx, cancel := context.WithCancelCause(context.Background()) - v.Context = ctx - v.Cancel = cancel - v.Nylons = make([]atomic.Pointer[core.Nylon], len(v.Central.Routers)) - v.nodeDone = make([]chan struct{}, len(v.Central.Routers)) - errChan := make(chan error, 128) // a large number so we dont get blocked - vn := &InMemoryNetwork{} - v.Net = vn - vn.cfg = v - nodes := len(v.Central.Routers) - vn.virtTun = make([]*tuntest.ChannelTUN, nodes) - vn.binds = make([]conn.Bind, nodes) - vn.readyCond = sync.NewCond(&sync.Mutex{}) - // pick the first endpoint specified for each node - vn.EpOutMapping = func(curNode state.NodeId, to bindtest.ChannelEndpoint2) bindtest.ChannelEndpoint2 { - for k, x := range v.Endpoints { - if x == curNode { - return bindtest.ChannelEndpoint2(netip.MustParseAddrPort(k)) + nd.nylon.Store(n) + h.launch(nd, n, 0) +} + +// Stop shuts every node and harness goroutine down. It runs on test cleanup +// and is safe to call more than once. +func (h *Harness) Stop() { + h.stop.Do(func() { + if h.cancel == nil { + return // never started + } + h.cancel(errors.New("harness stopped")) + for _, nd := range h.nodes { + if n := nd.nylon.Load(); n != nil { + n.Stop() } } - panic(fmt.Sprintf("no endpoint found for node %v", curNode)) - } - for e, n := range v.Endpoints { - idx := v.IndexOf(n) - v.Central.Routers[idx].Endpoints = append(v.Central.Routers[idx].Endpoints, e) - } - if v.LogLevel == nil { - v.LogLevel = new(slog.LevelDebug) - } - for idx := range v.Central.Routers { - n, err := core.NewNylon(v.Central, v.Local[idx], *v.LogLevel, "", map[string]any{ - "vnet": vn, - }, state.NylonOptions{DBG_log_wireguard: true}, v.Tunables) - if err != nil { - errChan <- err - return errChan + for _, nd := range h.nodes { + if nd.done != nil { + <-nd.done + } } - v.Nylons[idx].Store(n) - } - v.Net.Ready() - - startDelay := time.Duration(0) - for idx, rt := range v.Central.Routers { - sd := startDelay - n := v.Nylons[idx].Load() - done := make(chan struct{}) - v.nodeDone[idx] = done - go func() { - defer close(done) - timer := time.NewTimer(sd) - select { - case <-timer.C: - case <-ctx.Done(): - if !timer.Stop() { - select { - case <-timer.C: - default: - } - } + // Start's exit path cleans up a node, but a node that was built and + // never launched (Start failing partway) needs it here. Cleanup is + // idempotent. + for _, nd := range h.nodes { + if n := nd.nylon.Load(); n != nil { + _ = n.Cleanup() } - labels := pprof.Labels("nylon node", string(rt.Id)) - pprof.Do(context.Background(), labels, func(_ context.Context) { - if err := n.Start(); err != nil { - errChan <- err - } - }) - }() - startDelay += 500 * time.Millisecond - } - return errChan + } + h.mu.Lock() + for _, b := range h.binds { + if b != nil { + _ = b.Close() + } + } + h.mu.Unlock() + h.wg.Wait() + }) } -func (v *VirtualHarness) Stop() { - println("Stopping VirtualHarness") - v.Cancel(fmt.Errorf("stopping harness")) - for idx, _ := range v.Central.Routers { - v.Nylons[idx].Load().Stop() - } - v.Net.Stop() - println("Stopped VirtualHarness") +// fail reports a harness error and stops waits in progress. It is safe to +// call from any goroutine the harness owns. +func (h *Harness) fail(format string, args ...any) { + err := fmt.Errorf(format, args...) + h.t.Errorf("harness: %v", err) + h.cancel(err) } -type PacketFilter func(node state.NodeId, src, dst netip.Addr, data []byte) bool // return true to intercept, and stop the rest of the stack -func (h PacketFilter) TryApply(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if h == nil { - return false +// spawn runs fn on a goroutine that Stop waits for. +func (h *Harness) spawn(fn func()) { + h.wg.Add(1) + go func() { + defer h.wg.Done() + fn() + }() +} + +// ---- waiting ---- + +// Eventually polls cond until it returns true, failing the test after timeout. +func (h *Harness) Eventually(timeout time.Duration, what string, cond func() bool) { + h.t.Helper() + deadline := time.Now().Add(timeout) + for !cond() { + if time.Now().After(deadline) { + h.t.Fatalf("timed out after %s waiting for %s", timeout, what) + } + select { + case <-h.ctx.Done(): + h.t.Fatalf("harness stopped while waiting for %s: %v", what, context.Cause(h.ctx)) + case <-time.After(pollInterval): + } } - return h(node, src, dst, data) } -type OutMapping func(curNode state.NodeId, to bindtest.ChannelEndpoint2) bindtest.ChannelEndpoint2 +// await receives from ch, failing the test after timeout. +func await[T any](h *Harness, ch <-chan T, timeout time.Duration, what string) T { + h.t.Helper() + select { + case v := <-ch: + return v + case <-time.After(timeout): + h.t.Fatalf("timed out after %s waiting for %s", timeout, what) + case <-h.ctx.Done(): + h.t.Fatalf("harness stopped while waiting for %s: %v", what, context.Cause(h.ctx)) + } + panic("unreachable") +} -type InMemoryNetwork struct { - sync.Mutex - cfg *VirtualHarness - binds []conn.Bind - virtTun []*tuntest.ChannelTUN - SelfHandler PacketFilter // packet filter for handling packets destined for the current node - TransitHandler PacketFilter // packet filter for handling packets passing through the current node - EpOutMapping OutMapping - ready atomic.Bool - readyCond *sync.Cond +// OnNode runs fn on node id's dispatcher and returns its result. The +// dispatcher owns router state, so fn may read and change it freely. +func OnNode[T any](h *Harness, id state.NodeId, fn func(n *core.Nylon) T) T { + h.t.Helper() + n := h.Nylon(id) + res := make(chan T, 1) + n.Dispatch(func() error { + res <- fn(n) + return nil + }) + return await(h, res, dispatchTimeout, "dispatch on node "+string(id)) } -func (i *InMemoryNetwork) WaitForReady() { - i.readyCond.L.Lock() - defer i.readyCond.L.Unlock() - for !i.ready.Load() { - i.readyCond.Wait() +// ApplyConfig applies cfg on node id and fails the test unless it applied. +func (h *Harness) ApplyConfig(id state.NodeId, cfg *state.CentralCfg) { + h.t.Helper() + type result struct { + res core.ApplyResult + err error + } + r := OnNode(h, id, func(n *core.Nylon) result { + res, err := n.ApplyCentralConfig(cfg) + return result{res, err} + }) + if r.err != nil { + h.t.Fatalf("apply config on %s: %v", id, r.err) + } + if r.res != core.ApplyApplied { + h.t.Fatalf("apply config on %s: got result %v, want applied", id, r.res) } } -func (i *InMemoryNetwork) Ready() { - i.Lock() - defer i.Unlock() - i.ready.Store(true) - i.readyCond.Broadcast() +// ---- packets ---- + +// PacketFilter sees an IPv4 packet at node. Returning true consumes it. +type PacketFilter func(node state.NodeId, src, dst netip.Addr, data []byte) bool + +// OnDeliver sets the filter for packets that reach the node they are addressed +// to. Without one, delivered packets are dropped. +func (h *Harness) OnDeliver(f PacketFilter) { h.onDeliver.Store(&f) } + +// OnTransit sets the filter for packets a node forwards on to another node. +// Packets it does not consume are routed onwards as usual. +func (h *Harness) OnTransit(f PacketFilter) { h.onTransit.Store(&f) } + +func applyFilter(p *atomic.Pointer[PacketFilter], node state.NodeId, src, dst netip.Addr, data []byte) bool { + f := p.Load() + return f != nil && (*f)(node, src, dst, data) } -func (i *InMemoryNetwork) virtualRouteTable(node state.NodeId, src, dst netip.Addr, data []byte, pkt []byte) bool { - curCfg := i.cfg.Central.GetNode(node) - if pkt[8] == 0 { // handle self if ttl is 0 as well - if i.SelfHandler.TryApply(node, src, dst, data) { - return true - } +// Send injects an IPv4 packet carrying payload into node's TUN, as if an +// application on that node had sent it. +func (h *Harness) Send(node state.NodeId, src, dst string, payload []byte, ttl byte) { + pkt := make([]byte, ipv4HeaderLen, ipv4HeaderLen+len(payload)) + pkt[0] = 4<<4 | ipv4HeaderLen/4 + binary.BigEndian.PutUint16(pkt[2:4], uint16(ipv4HeaderLen+len(payload))) + pkt[8] = ttl + copy(pkt[12:16], netip.MustParseAddr(src).AsSlice()) + copy(pkt[16:20], netip.MustParseAddr(dst).AsSlice()) + pkt = append(pkt, payload...) + + h.mu.RLock() + tn := h.tuns[h.indexOf(node)] + h.mu.RUnlock() + select { + case tn.Outbound <- pkt: + case <-h.ctx.Done(): } - for _, prefix := range curCfg.Prefixes { - if prefix.GetPrefix().Contains(dst) { - if i.SelfHandler.TryApply(node, src, dst, data) { - return true +} + +// SendEvery sends the packet every interval until the harness stops or the +// returned function is called. +func (h *Harness) SendEvery(interval time.Duration, node state.NodeId, src, dst string, payload []byte, ttl byte) (stop func()) { + ctx, cancel := context.WithCancel(h.ctx) + h.spawn(func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + h.Send(node, src, dst, payload, ttl) } } + }) + return cancel +} + +// ---- links ---- + +type linkKey struct{ from, to netip.AddrPort } + +// Link is a one-way path between two endpoints. Packets on endpoint pairs +// without a link are dropped. +type Link struct { + mu sync.Mutex + latency time.Duration + jitter time.Duration + packetLoss float64 +} + +// Link adds a one-way link from one endpoint to another. +func (h *Harness) Link(from, to string) *Link { + l := &Link{} + h.mu.Lock() + h.links[linkKey{netip.MustParseAddrPort(from), netip.MustParseAddrPort(to)}] = l + h.mu.Unlock() + return l +} + +// Links is a set of links configured together. +type Links []*Link + +// Connect adds links in both directions between two endpoints. +func (h *Harness) Connect(a, b string) Links { + return Links{h.Link(a, b), h.Link(b, a)} +} + +// WithLatency delays each packet by lat plus up to jitter. +func (l *Link) WithLatency(lat, jitter time.Duration) *Link { + l.mu.Lock() + defer l.mu.Unlock() + l.latency, l.jitter = lat, jitter + return l +} + +// WithPacketLoss drops each packet with probability loss. +func (l *Link) WithPacketLoss(loss float64) *Link { + l.mu.Lock() + defer l.mu.Unlock() + l.packetLoss = loss + return l +} + +func (ls Links) WithLatency(lat, jitter time.Duration) Links { + for _, l := range ls { + l.WithLatency(lat, jitter) } - curIdx := i.cfg.IndexOf(node) + return ls +} - if !i.TransitHandler.TryApply(node, src, dst, data) { - // default routing behaviour - for _, n := range i.cfg.Central.Routers { - if node == n.Id { - continue - } - for _, prefix := range n.Prefixes { - if prefix.GetPrefix().Contains(dst) { - // route to this node - select { - case i.virtTun[curIdx].Outbound <- pkt: // send back into our tun to get routed by WireGuard/Polyamide - return true - default: - fmt.Printf("%s's tun is not ready to accept data\n", n.Id) - return true - } - } - } - } - return false +func (ls Links) WithPacketLoss(loss float64) Links { + for _, l := range ls { + l.WithPacketLoss(loss) } - return true + return ls +} + +func (h *Harness) random() float64 { + h.rngMu.Lock() + defer h.rngMu.Unlock() + return h.rng.Float64() } -func (i *InMemoryNetwork) virtualInternet(pkt []byte, len int, from, to bindtest.ChannelEndpoint2) { - // simulate network conditions - i.cfg.linksMu.RLock() - idx := slices.IndexFunc(i.cfg.Links, func(link *VirtualLink) bool { - return link.Edge.V1 == from && link.Edge.V2 == to +// transmit carries a packet across the link from one endpoint to another. +func (h *Harness) transmit(pkt []byte, from, to netip.AddrPort) { + h.mu.RLock() + link := h.links[linkKey{from, to}] + h.mu.RUnlock() + if link == nil { + return // no link, the packet is lost + } + link.mu.Lock() + latency, jitter, loss := link.latency, link.jitter, link.packetLoss + link.mu.Unlock() + + if loss > 0 && h.random() < loss { + return + } + if latency == 0 && jitter == 0 { + h.receive(pkt, from, to) + return + } + delay := latency + time.Duration(h.random()*float64(jitter)) + h.spawn(func() { + select { + case <-h.ctx.Done(): + case <-time.After(delay): + h.receive(pkt, from, to) + } }) - var link *VirtualLink - if idx != -1 { - link = i.cfg.Links[idx] +} + +// receive hands a packet to the bind of the node that owns endpoint to. +func (h *Harness) receive(pkt []byte, from, to netip.AddrPort) { + owner, ok := h.epOwner[to] + if !ok { + return } - i.cfg.linksMu.RUnlock() - if link == nil { - return // no connection, dropped packet + h.mu.RLock() + b := h.binds[h.indexOf(owner)] + h.mu.RUnlock() + if b == nil { + return // node is restarting + } + err := b.Send([][]byte{pkt}, bindtest.ChannelEndpoint2(from)) + if err != nil && !errors.Is(err, net.ErrClosed) { + h.fail("deliver packet to %s: %v", owner, err) } - link.simulate(pkt, len, from, to, i) } -func (i *InMemoryNetwork) NewRuntime(n *core.Nylon) (core.Runtime, error) { +// ---- runtime for core ---- + +// NewRuntime gives a node a virtual TUN and a WireGuard transport whose UDP +// socket is attached to the simulated network. core calls it through the +// "vnet" aux config entry. +func (h *Harness) NewRuntime(n *core.Nylon) (core.Runtime, error) { + id := n.LocalCfg.Id var host tun.Device if n.NoTun { host = tun.NewDummyDevice("nylon-vn") } else { - host = i.Tun(n.LocalCfg.Id) + host = h.newTun(id) } - transport, err := core.NewWireGuardTransport(n, func() conn.Bind { return i.newWireGuardBind(n.LocalCfg.Id) }) + transport, err := core.NewWireGuardTransport(n, func() conn.Bind { return h.newBind(id) }) if err != nil { _ = host.Close() return core.Runtime{}, err @@ -355,117 +582,141 @@ func (i *InMemoryNetwork) NewRuntime(n *core.Nylon) (core.Runtime, error) { return core.Runtime{Transports: []polyamide.Transport{transport}, Host: host}, nil } -func (i *InMemoryNetwork) newWireGuardBind(node state.NodeId) conn.Bind { - i.Lock() - defer i.Unlock() - epSendMapping := func(to bindtest.ChannelEndpoint2) bindtest.ChannelEndpoint2 { - return i.EpOutMapping(node, to) - } - bp := bindtest.NewChannelBind2() - - remote := bp[0] // simulate an actual port opening - open, _, err := remote.Open(0) +// newBind returns the node's end of a channel bind, and pumps packets the +// node sends from the other end onto the simulated network. +func (h *Harness) newBind(id state.NodeId) conn.Bind { + pair := bindtest.NewChannelBind2() + remote := pair[0] + recvFns, _, err := remote.Open(0) if err != nil { + h.fail("open bind for %s: %v", id, err) return nil } - numId := i.cfg.IndexOf(node) - i.binds[numId] = remote - go func() { - // bind listener routine for packets sent from this node - bufSize := remote.BatchSize() - pktBuf := make([][]byte, bufSize) - pktBuf[0] = make([]byte, device.MaxMessageSize) - lenBuf := make([]int, bufSize) - epBuf := make([]conn.Endpoint, bufSize) - i.WaitForReady() + h.mu.Lock() + h.binds[h.indexOf(id)] = remote + h.mu.Unlock() + from := h.firstEp[id] + + h.spawn(func() { + select { + case <-h.ready: + case <-h.ctx.Done(): + return + } + batch := remote.BatchSize() + bufs := make([][]byte, batch) + bufs[0] = make([]byte, device.MaxMessageSize) + sizes := make([]int, batch) + eps := make([]conn.Endpoint, batch) for { - for _, recv := range open { - n, err := recv(pktBuf, lenBuf, epBuf) + for _, recv := range recvFns { + n, err := recv(bufs, sizes, eps) if err != nil { if !errors.Is(err, net.ErrClosed) { - panic(err) + h.fail("receive from %s's bind: %v", id, err) } return } - for pi := range n { - if lenBuf[pi] == 0 { + for i := range n { + if sizes[i] == 0 { continue } - toIp := epBuf[pi].(bindtest.ChannelEndpoint2) - fromIp := epSendMapping(toIp) - i.virtualInternet(slices.Clone(pktBuf[pi]), lenBuf[pi], fromIp, toIp) + to := netip.AddrPort(eps[i].(bindtest.ChannelEndpoint2)) + h.transmit(slices.Clone(bufs[i][:sizes[i]]), from, to) } } } - }() - return bp[1] -} - -func (i *InMemoryNetwork) Tun(node state.NodeId) tun.Device { - i.Lock() - defer i.Unlock() - const ( - ipv4Size = 20 - ) - bt := tuntest.NewChannelTUN() - - numId := slices.IndexFunc(i.cfg.Central.Routers, func(cfg state.RouterCfg) bool { - return cfg.Id == node }) + return pair[1] +} - i.virtTun[numId] = bt - go func() { - i.WaitForReady() +// newTun returns a TUN for node id. Packets the node writes to it are either +// delivered locally or, when addressed to another node, sent back into the +// TUN so the node routes them like any outbound packet. +func (h *Harness) newTun(id state.NodeId) tun.Device { + tn := tuntest.NewChannelTUN() + h.mu.Lock() + h.tuns[h.indexOf(id)] = tn + h.mu.Unlock() + + h.spawn(func() { + select { + case <-h.ready: + case <-h.ctx.Done(): + return + } for { select { - case <-i.cfg.Context.Done(): + case <-h.ctx.Done(): return - case pkt := <-bt.Inbound: - // wireguard doesn't actually verify checksum for ip - ip := pkt[0:ipv4Size] + case pkt := <-tn.Inbound: if pkt[0]>>4 != 4 { - panic("unexpected packet, not ipv4") - } - var src, dst netip.Addr - err := src.UnmarshalBinary(ip[12:16]) - if err != nil { - panic(err) - } - err = dst.UnmarshalBinary(ip[16:20]) - if err != nil { - panic(err) + h.fail("node %s wrote a non-IPv4 packet to its TUN", id) + return } - if !i.virtualRouteTable(node, src, dst, pkt[ipv4Size:], pkt) { - panic(fmt.Sprintf("unhandled packet src: %v, dst: %v", src, dst)) + src, _ := netip.AddrFromSlice(pkt[12:16]) + dst, _ := netip.AddrFromSlice(pkt[16:20]) + if !h.routeFromTun(id, tn, src, dst, pkt) { + h.fail("node %s: no route for packet %v -> %v", id, src, dst) + return } } } - }() - return bt.TUN() -} - -func (i *InMemoryNetwork) Send(node state.NodeId, src, dst string, pkt []byte, ttl byte) { - const ( - ipv4Size = 20 - ) - numId := i.cfg.IndexOf(node) - ipPkt := append(make([]byte, ipv4Size), pkt...) - ip := ipPkt[0:ipv4Size] - ip[0] = 4<<4 | ipv4Size/4 - binary.BigEndian.PutUint16(ip[2:4], uint16(ipv4Size+len(pkt))) - ip[8] = ttl - copy(ipPkt[12:16], netip.MustParseAddr(src).AsSlice()) - copy(ipPkt[16:20], netip.MustParseAddr(dst).AsSlice()) - i.virtTun[numId].Outbound <- ipPkt -} - -func (i *InMemoryNetwork) Stop() { - for _, d := range i.binds { - if d != nil { - err := d.Close() - if err != nil { - panic(err) + }) + return tn.TUN() +} + +// routeFromTun stands in for the host's routing table. +func (h *Harness) routeFromTun(id state.NodeId, tn *tuntest.ChannelTUN, src, dst netip.Addr, pkt []byte) bool { + data := pkt[ipv4HeaderLen:] + // a packet whose TTL ran out is handed to the local stack as well + if pkt[8] == 0 && applyFilter(&h.onDeliver, id, src, dst, data) { + return true + } + for _, prefix := range h.Central.GetNode(id).Prefixes { + if prefix.GetPrefix().Contains(dst) && applyFilter(&h.onDeliver, id, src, dst, data) { + return true + } + } + if applyFilter(&h.onTransit, id, src, dst, data) { + return true + } + for _, router := range h.Central.Routers { + if router.Id == id { + continue + } + for _, prefix := range router.Prefixes { + if prefix.GetPrefix().Contains(dst) { + select { + case tn.Outbound <- pkt: + default: + h.t.Logf("node %s: TUN outbound queue full, dropping packet", id) + } + return true } } } + return false +} + +// Signal is a one-shot event that can be triggered more than once. +type Signal chan struct{} + +func NewSignal() Signal { return make(Signal) } + +func (s Signal) Trigger() { + select { + case <-s: + default: + close(s) + } +} + +func (s Signal) Triggered() bool { + select { + case <-s: + return true + default: + return false + } } diff --git a/integration/ipc_test.go b/integration/ipc_test.go index de4b78b5..170f735c 100644 --- a/integration/ipc_test.go +++ b/integration/ipc_test.go @@ -7,9 +7,11 @@ import ( "errors" "io" "net" + "net/netip" "os" "sync" "testing" + "testing/synctest" "time" "github.com/encodeous/nylon/core" @@ -19,25 +21,32 @@ import ( "github.com/goccy/go-yaml" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "go.uber.org/goleak" ) -func setupTwoNodeHarness(t *testing.T) (*VirtualHarness, chan error) { +// startTwoNodes runs a - b, where b has two endpoints. +func startTwoNodes(t *testing.T) *Harness { t.Helper() - vh := &VirtualHarness{} - a1 := "192.168.50.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.50.2:1234" - b2 := "192.168.50.3:1234" - vh.NewNode("b", "10.0.0.2/32") - vh.Central.Graph = []string{"a, b"} - vh.Endpoints = map[string]state.NodeId{a1: "a", b1: "b", b2: "b"} - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(a1, b2) - vh.AddLink(b2, a1) - errs := vh.Start() - return vh, errs + h := NewHarness(t) + a1, b1, b2 := "192.168.50.1:1234", "192.168.50.2:1234", "192.168.50.3:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1, b2) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + h.Connect(a1, b2) + h.Start() + return h +} + +// startUnreachablePair runs a and b with every packet between them lost. +func startUnreachablePair(t *testing.T, a1, b1 string) *Harness { + t.Helper() + h := NewHarness(t) + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1).WithPacketLoss(1) + h.Start() + return h } // ipcClient connects a client to n's IPC server over an in-memory pipe. @@ -71,243 +80,202 @@ func ipcErrorCode(t *testing.T, err error) protocol.IpcError_Code { return ipcErr.Code } -func TestIPCStatus(t *testing.T) { - defer goleak.VerifyNone(t) - vh, _ := setupTwoNodeHarness(t) - defer vh.Stop() +// rawIPCConn serves n's IPC on a pipe and returns the client end, closed when +// the test ends. +func rawIPCConn(t *testing.T, n *core.Nylon) (net.Conn, <-chan error) { + t.Helper() + client, server := net.Pipe() + served := make(chan error, 1) + done := make(chan struct{}) + go func() { + defer close(done) + served <- core.ServeIPC(n, server) + }() + t.Cleanup(func() { + _ = client.Close() + <-done + }) + return client, served +} - // Wait for links to come up - time.Sleep(3 * time.Second) +func TestIPCStatus(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + bPrefix := netip.MustParsePrefix("10.0.0.2/32") + h.Eventually(10*time.Second, "a to select a route to b", func() bool { + return OnNode(h, "a", func(n *core.Nylon) bool { + route, ok := n.RouterState.Routes[bPrefix] + return ok && route.Metric != state.INF + }) + }) - a := vh.Nylons[vh.IndexOf("a")].Load() - resp, err := ipcCall(t, a, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}, + resp, err := ipcCall(t, h.Nylon("a"), &protocol.IpcRequest{ + Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}, + }) + require.NoError(t, err) + s := resp.GetStatus() + require.NotNil(t, s.GetNode()) + assert.Equal(t, "a", s.GetNode().NodeId) + assert.NotEmpty(t, s.GetNode().PublicKey) + assert.Equal(t, int32(1), s.GetNode().GetStats().NeighbourCount) + assert.GreaterOrEqual(t, s.GetNode().GetStats().SelectedRouteCount, int32(1)) + assert.GreaterOrEqual(t, s.GetNode().GetStats().AdvertisedPrefixCount, int32(1)) + + require.Len(t, s.GetNeighbours(), 1) + peer := s.GetNeighbours()[0] + assert.Equal(t, "b", peer.PeerId) + assert.NotEmpty(t, peer.PublicKey) + require.NotEmpty(t, peer.GetEndpoints()) + assert.GreaterOrEqual(t, len(peer.GetEndpoints()), 2) + assert.NotEmpty(t, peer.GetEndpoints()[0].Address) + + assert.GreaterOrEqual(t, len(s.GetRoutes().GetSelected()), 1) + assert.GreaterOrEqual(t, len(s.GetRoutes().GetForward()), 1) + assert.GreaterOrEqual(t, len(s.GetFeasibilityDistances()), 1) }) - require.NoError(t, err) - s := resp.GetStatus() - require.NotNil(t, s.GetNode()) - assert.Equal(t, "a", s.GetNode().NodeId) - assert.NotEmpty(t, s.GetNode().PublicKey) - assert.Equal(t, int32(1), s.GetNode().GetStats().NeighbourCount) - assert.GreaterOrEqual(t, s.GetNode().GetStats().SelectedRouteCount, int32(1)) - assert.GreaterOrEqual(t, s.GetNode().GetStats().AdvertisedPrefixCount, int32(1)) - - require.Len(t, s.GetNeighbours(), 1) - peer := s.GetNeighbours()[0] - assert.Equal(t, "b", peer.PeerId) - assert.NotEmpty(t, peer.PublicKey) - require.NotEmpty(t, peer.GetEndpoints()) - assert.GreaterOrEqual(t, len(peer.GetEndpoints()), 2) - assert.NotEmpty(t, peer.GetEndpoints()[0].Address) - - assert.GreaterOrEqual(t, len(s.GetRoutes().GetSelected()), 1) - assert.GreaterOrEqual(t, len(s.GetRoutes().GetForward()), 1) - 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, err := ipcCall(t, a, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 100}}, + synctest.Test(t, func(t *testing.T) { + h := startUnreachablePair(t, "192.168.52.1:1234", "192.168.52.2:1234") + time.Sleep(500 * time.Millisecond) // let both nodes start + + resp, err := ipcCall(t, h.Nylon("a"), &protocol.IpcRequest{ + Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 100}}, + }) + require.NoError(t, err) + 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) + } }) - require.NoError(t, err) - 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) - defer vh.Stop() - - time.Sleep(1 * time.Second) - - // Write the current central config to a temp file - cfgData, err := yaml.Marshal(&vh.Central) - require.NoError(t, err) - tmpFile := t.TempDir() + "/central.yaml" - require.NoError(t, os.WriteFile(tmpFile, cfgData, 0600)) - - a := vh.Nylons[vh.IndexOf("a")].Load() - a.ConfigPath = tmpFile - client := ipcClient(t, a) - type result struct { - resp *protocol.IpcResponse - err error - } - done := make(chan result, 1) - go func() { - resp, err := client.Do(&protocol.IpcRequest{ + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + time.Sleep(time.Second) // let both nodes start + + // Write the current central config to a temp file + cfgData, err := yaml.Marshal(&h.Central) + require.NoError(t, err) + tmpFile := t.TempDir() + "/central.yaml" + require.NoError(t, os.WriteFile(tmpFile, cfgData, 0600)) + + a := h.Nylon("a") + a.ConfigPath = tmpFile + resp, err := ipcCall(t, a, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Reload{Reload: &protocol.ReloadRequest{}}, }) - done <- result{resp, err} - }() - - select { - case res := <-done: - require.NoError(t, res.err) - r := res.resp.GetReload() - assert.Equal(t, protocol.ReloadResult_NOOP, r.Result) - case err := <-errs: - t.Fatal(err) - case <-time.After(10 * time.Second): - t.Fatal("timeout") - } + require.NoError(t, err) + assert.Equal(t, protocol.ReloadResult_NOOP, resp.GetReload().Result) + }) } func TestIPCProbeNonNeighbour(t *testing.T) { - defer goleak.VerifyNone(t) - vh, _ := setupTwoNodeHarness(t) - defer vh.Stop() - - time.Sleep(1 * time.Second) + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + time.Sleep(time.Second) // let both nodes start - a := vh.Nylons[vh.IndexOf("a")].Load() - _, err := ipcCall(t, a, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "nonexistent"}}, + _, err := ipcCall(t, h.Nylon("a"), &protocol.IpcRequest{ + Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "nonexistent"}}, + }) + assert.Equal(t, protocol.IpcError_NOT_FOUND, ipcErrorCode(t, err)) + assert.Contains(t, err.Error(), "not a neighbour") }) - assert.Equal(t, protocol.IpcError_NOT_FOUND, ipcErrorCode(t, err)) - assert.Contains(t, err.Error(), "not a neighbour") } func TestIPCTraceDisabled(t *testing.T) { - defer goleak.VerifyNone(t) - vh, _ := setupTwoNodeHarness(t) - defer vh.Stop() - - time.Sleep(1 * time.Second) + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + time.Sleep(time.Second) // let both nodes start - a := vh.Nylons[vh.IndexOf("a")].Load() - // DBG_trace_tc is false by default - _, err := ipcCall(t, a, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, + // DBG_trace_tc is false by default + _, err := ipcCall(t, h.Nylon("a"), &protocol.IpcRequest{ + Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, + }) + assert.Equal(t, protocol.IpcError_FAILED_PRECONDITION, ipcErrorCode(t, err)) + assert.Contains(t, err.Error(), "tracing not enabled") }) - assert.Equal(t, protocol.IpcError_FAILED_PRECONDITION, ipcErrorCode(t, err)) - assert.Contains(t, err.Error(), "tracing not enabled") } // A frame that isn't an IpcRequest gets an error response, then the server hangs up. func TestIPCMalformedRequest(t *testing.T) { - defer goleak.VerifyNone(t) - vh, _ := setupTwoNodeHarness(t) - defer vh.Stop() + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + time.Sleep(time.Second) // let both nodes start - time.Sleep(1 * time.Second) + client, _ := rawIPCConn(t, h.Nylon("a")) + require.NoError(t, ipc.WriteHello(client)) + version, err := ipc.ReadHello(client) + require.NoError(t, err) + require.Equal(t, ipc.Version, version) - a := vh.Nylons[vh.IndexOf("a")].Load() - client, server := net.Pipe() - defer client.Close() - go core.ServeIPC(a, server) - - require.NoError(t, ipc.WriteHello(client)) - version, err := ipc.ReadHello(client) - require.NoError(t, err) - require.Equal(t, ipc.Version, version) + garbage := []byte{0xff, 0xff, 0xff} + frame := binary.BigEndian.AppendUint32(nil, uint32(len(garbage))) + _, err = client.Write(append(frame, garbage...)) + require.NoError(t, err) - garbage := []byte{0xff, 0xff, 0xff} - frame := binary.BigEndian.AppendUint32(nil, uint32(len(garbage))) - _, err = client.Write(append(frame, garbage...)) - require.NoError(t, err) - - resp := &protocol.IpcResponse{} - require.NoError(t, ipc.ReadFrame(client, resp)) - assert.Equal(t, protocol.IpcError_INVALID_REQUEST, resp.GetError().GetCode()) - _, err = client.Read(make([]byte, 1)) - assert.ErrorIs(t, err, io.EOF) + resp := &protocol.IpcResponse{} + require.NoError(t, ipc.ReadFrame(client, resp)) + assert.Equal(t, protocol.IpcError_INVALID_REQUEST, resp.GetError().GetCode()) + _, err = client.Read(make([]byte, 1)) + assert.ErrorIs(t, err, io.EOF) + }) } // A client with another protocol version learns the server's, and the server hangs up. func TestIPCVersionMismatch(t *testing.T) { - defer goleak.VerifyNone(t) - vh, _ := setupTwoNodeHarness(t) - defer vh.Stop() - - a := vh.Nylons[vh.IndexOf("a")].Load() - client, server := net.Pipe() - defer client.Close() - served := make(chan error, 1) - go func() { served <- core.ServeIPC(a, server) }() - - _, err := client.Write([]byte{'N', 'Y', 'L', 'N', 0xff, 0xff}) - require.NoError(t, err) - version, err := ipc.ReadHello(client) - require.NoError(t, err) - assert.Equal(t, ipc.Version, version) - assert.ErrorAs(t, <-served, &ipc.ErrVersion{}) + synctest.Test(t, func(t *testing.T) { + h := startTwoNodes(t) + + client, served := rawIPCConn(t, h.Nylon("a")) + _, err := client.Write([]byte{'N', 'Y', 'L', 'N', 0xff, 0xff}) + require.NoError(t, err) + version, err := ipc.ReadHello(client) + require.NoError(t, err) + assert.Equal(t, ipc.Version, version) + err = <-served + assert.ErrorAs(t, err, &ipc.ErrVersion{}) + }) } // Requests on one connection run concurrently, and cancelling one ends it early. func TestIPCConcurrentRequestsAndCancel(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - a1 := "192.168.53.1:1234" - b1 := "192.168.53.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) - vh.Start() - defer vh.Stop() - - time.Sleep(500 * time.Millisecond) - - a := vh.Nylons[vh.IndexOf("a")].Load() - client, server := net.Pipe() - defer client.Close() - go core.ServeIPC(a, server) - require.NoError(t, ipc.WriteHello(client)) - _, err := ipc.ReadHello(client) - require.NoError(t, err) - require.NoError(t, client.SetDeadline(time.Now().Add(10*time.Second))) - - var writeMu sync.Mutex - send := func(req *protocol.IpcRequest) { - writeMu.Lock() - defer writeMu.Unlock() - require.NoError(t, ipc.WriteFrame(client, req)) - } - read := func() *protocol.IpcResponse { - resp := &protocol.IpcResponse{} - require.NoError(t, ipc.ReadFrame(client, resp)) - return resp - } - // Every packet is lost, so the probe waits out its whole timeout. - start := time.Now() - send(&protocol.IpcRequest{Id: 1, Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 5000}}}) - send(&protocol.IpcRequest{Id: 2, Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}}) - resp := read() - require.Equal(t, uint32(2), resp.Id, "status should not wait for the probe") - require.Equal(t, "a", resp.GetStatus().GetNode().GetNodeId()) - - send(&protocol.IpcRequest{Request: &protocol.IpcRequest_Cancel{Cancel: &protocol.CancelRequest{Id: 1}}}) - resp = read() - require.Equal(t, uint32(1), resp.Id) - require.Less(t, time.Since(start), 4*time.Second, "the cancelled probe should end before its timeout") + synctest.Test(t, func(t *testing.T) { + h := startUnreachablePair(t, "192.168.53.1:1234", "192.168.53.2:1234") + time.Sleep(500 * time.Millisecond) // let both nodes start + + client, _ := rawIPCConn(t, h.Nylon("a")) + require.NoError(t, ipc.WriteHello(client)) + _, err := ipc.ReadHello(client) + require.NoError(t, err) + require.NoError(t, client.SetDeadline(time.Now().Add(10*time.Second))) + + var writeMu sync.Mutex + send := func(req *protocol.IpcRequest) { + writeMu.Lock() + defer writeMu.Unlock() + require.NoError(t, ipc.WriteFrame(client, req)) + } + read := func() *protocol.IpcResponse { + resp := &protocol.IpcResponse{} + require.NoError(t, ipc.ReadFrame(client, resp)) + return resp + } + // Every packet is lost, so the probe waits out its whole timeout. + start := time.Now() + send(&protocol.IpcRequest{Id: 1, Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 5000}}}) + send(&protocol.IpcRequest{Id: 2, Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}}) + resp := read() + require.Equal(t, uint32(2), resp.Id, "status should not wait for the probe") + require.Equal(t, "a", resp.GetStatus().GetNode().GetNodeId()) + + send(&protocol.IpcRequest{Request: &protocol.IpcRequest_Cancel{Cancel: &protocol.CancelRequest{Id: 1}}}) + resp = read() + require.Equal(t, uint32(1), resp.Id) + require.Less(t, time.Since(start), 4*time.Second, "the cancelled probe should end before its timeout") + }) } diff --git a/integration/race_test.go b/integration/race_test.go index 9474ca6e..50c9e77e 100644 --- a/integration/race_test.go +++ b/integration/race_test.go @@ -7,103 +7,59 @@ import ( "log/slog" "net/netip" "testing" + "testing/synctest" "time" + "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/state" - "go.uber.org/goleak" ) +// TestRapidToggleConfig flips a's config between two versions while packets +// flow, so config applies race with ForwardTable lookups in the TC filter. func TestRapidToggleConfig(t *testing.T) { - defer goleak.VerifyNone(t) + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.LogLevel = slog.LevelInfo + a1, b1 := "192.168.1.1:1234", "192.168.1.2:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.Central.Graph = []string{"a, b"} + h.Connect(a1, b1) + h.OnDeliver(func(state.NodeId, netip.Addr, netip.Addr, []byte) bool { return true }) + h.Start() - vh := &VirtualHarness{} - vh.LogLevel = new(slog.LevelInfo) - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - 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) - vh.AddLink(b1, a1) + // Let the initial convergence happen. + time.Sleep(3 * time.Second) - errs := vh.Start() - - vn := vh.Net - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - return true - } - - // Wait for initial convergence. - time.Sleep(3 * time.Second) - - _, baseCfg := vh.Central.Clone() - _, extraCfg := vh.Central.Clone() - - aIdx := vh.IndexOf("a") - extraRouter := extraCfg.Routers[aIdx] - for i := 0; i < 50; i++ { - extraRouter.Prefixes = append(extraRouter.Prefixes, state.PrefixHealthWrapper{ - PrefixHealth: &state.StaticPrefixHealth{ - Prefix: netip.MustParsePrefix(fmt.Sprintf("10.1.0.%d/32", i)), - Metric: 0, - }, - }) - } - extraCfg.Routers[aIdx] = extraRouter + _, baseCfg := h.Central.Clone() + _, extraCfg := h.Central.Clone() + extraRouter := routerOf(extraCfg, "a") + for i := range 50 { + extraRouter.Prefixes = append(extraRouter.Prefixes, state.PrefixHealthWrapper{ + PrefixHealth: &state.StaticPrefixHealth{ + Prefix: netip.MustParsePrefix(fmt.Sprintf("10.1.0.%d/32", i)), + }, + }) + } - a := vh.Nylons[vh.IndexOf("a")].Load() + h.SendEvery(10*time.Millisecond, "a", "10.0.0.1", "10.0.0.2", []byte{1}, 64) - // Send packets every 10ms to trigger ForwardTable.Lookup via TC filter. - stop := make(chan struct{}) - go func() { - ticker := time.NewTicker(10 * time.Millisecond) - defer ticker.Stop() - for { - select { - case <-stop: - return - case <-vh.Context.Done(): - return - case <-ticker.C: - vn.Send("a", "10.0.0.1", "10.0.0.2", []byte{1}, 64) + deadline := time.Now().Add(5 * time.Second) + for i := 0; time.Now().Before(deadline); i++ { + cfg := extraCfg + if i%2 == 1 { + cfg = baseCfg } - } - }() - - // Rapidly toggle config until test duration expires. - deadline := time.After(5 * time.Second) - for i := 0; ; i++ { - cfg := extraCfg - if i%2 == 1 { - cfg = baseCfg - } - cfg.Timestamp = baseCfg.Timestamp + int64(i) + 2 - done := make(chan struct{}) - _, ccfg := cfg.Clone() - a.Dispatch(func() error { - defer close(done) - _, err := a.ApplyCentralConfig(ccfg) - if err != nil { + cfg.Timestamp = baseCfg.Timestamp + int64(i) + 2 + _, next := cfg.Clone() + err := OnNode(h, "a", func(n *core.Nylon) error { + _, err := n.ApplyCentralConfig(next) return err + }) + if err != nil { + t.Fatalf("apply config #%d: %v", i, err) } - return nil - }) - select { - case <-done: time.Sleep(10 * time.Millisecond) - case <-deadline: - goto end - case err := <-errs: - t.Fatalf("harness error: %v", err) } - } -end: - close(stop) - vh.Stop() + }) } diff --git a/integration/routing_test.go b/integration/routing_test.go index 8f9e9d1a..dddca174 100644 --- a/integration/routing_test.go +++ b/integration/routing_test.go @@ -5,183 +5,69 @@ package integration import ( "net/netip" "testing" + "testing/synctest" "time" "github.com/encodeous/nylon/state" - "go.uber.org/goleak" ) -func TestInProcessRouting(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - vh.UntrackedRouting = true - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") - c1 := "192.168.1.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b", - "b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(b1, c1) - vh.AddLink(c1, b1) - - errs := vh.Start() - - vn := vh.Net - cc := make(chan bool, 100) - - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "c" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 { - cc <- true - } - return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vn.Send("a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) - } - } - }() +// lineTopology builds a - b - c with each node on its own endpoint. +func lineTopology(h *Harness) { + a1, b1, c1 := "192.168.1.1:1234", "192.168.1.2:1234", "192.168.1.3:1234" + h.AddNode("a", "10.0.0.1/32", a1) + h.AddNode("b", "10.0.0.2/32", b1) + h.AddNode("c", "10.0.0.3/32", c1) + h.Central.Graph = []string{"a, b", "b, c"} + h.Connect(a1, b1) + h.Connect(b1, c1) +} - select { - case <-cc: - t.Log("Got ping!") - case <-time.After(10 * time.Second): - t.Error("Timed out waiting for ping") - case err := <-errs: - t.Error(err) - } - vh.Stop() +func TestInProcessRouting(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.UntrackedRouting = true + lineTopology(h) + + got := expectPacket(h, "c", "10.0.0.1", "10.0.0.3", 222) + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) + await(h, got, 10*time.Second, "ping from a to c") + }) } func TestInProcessRoutingThroughTUNLessRelay(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - vh.UntrackedRouting = true - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") - vh.Local[vh.IndexOf("b")].NoTun = true - c1 := "192.168.1.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b", - "b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(b1, c1) - vh.AddLink(c1, b1) - - errs := vh.Start() - defer vh.Stop() - - received := make(chan struct{}, 1) - vh.Net.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if node == "c" && src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 { - received <- struct{}{} - } - return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vh.Net.Send("a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) - } - } - }() - - select { - case <-received: - case <-time.After(10 * time.Second): - t.Error("timed out waiting for packet through TUN-less relay") - case err := <-errs: - t.Error(err) - } + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.UntrackedRouting = true + lineTopology(h) + h.LocalCfg("b").NoTun = true + + got := expectPacket(h, "c", "10.0.0.1", "10.0.0.3", 222) + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.3", []byte{222}, 64) + await(h, got, 10*time.Second, "packet through TUN-less relay") + }) } func TestTTL(t *testing.T) { - defer goleak.VerifyNone(t) - vh := &VirtualHarness{} - vh.UntrackedRouting = true - a1 := "192.168.1.1:1234" - vh.NewNode("a", "10.0.0.1/32") - b1 := "192.168.1.2:1234" - vh.NewNode("b", "10.0.0.2/32") - c1 := "192.168.1.3:1234" - vh.NewNode("c", "10.0.0.3/32") - vh.Central.Graph = []string{ - "a, b", - "b, c", - } - vh.Endpoints = map[string]state.NodeId{ - a1: "a", - b1: "b", - c1: "c", - } - vh.AddLink(a1, b1) - vh.AddLink(b1, a1) - vh.AddLink(b1, c1) - vh.AddLink(c1, b1) - - errs := vh.Start() - - vn := vh.Net - cc := make(chan bool, 100) - - vn.SelfHandler = func(node state.NodeId, src, dst netip.Addr, data []byte) bool { - if src.String() == "10.0.0.1" && dst.String() == "10.0.0.3" && data[0] == 222 { - if node == "c" { - panic("c is not supposed to receive this packet") - } - cc <- true - } - return true - } - - go func() { - for { - select { - case <-vh.Context.Done(): - return - case <-time.After(100 * time.Millisecond): - vn.Send("a", "10.0.0.1", "10.0.0.3", []byte{222}, 1) + synctest.Test(t, func(t *testing.T) { + h := NewHarness(t) + h.UntrackedRouting = true + lineTopology(h) + + src, dst := netip.MustParseAddr("10.0.0.1"), netip.MustParseAddr("10.0.0.3") + expired := NewSignal() + h.OnDeliver(func(node state.NodeId, s, d netip.Addr, data []byte) bool { + if s == src && d == dst && data[0] == 222 { + if node == "c" { + t.Error("c received a packet whose TTL should have expired at b") + } + expired.Trigger() } - } - }() - - select { - case <-cc: - t.Log("Got ping!") - case <-time.After(10 * time.Second): - t.Error("Timed out waiting for ping") - case err := <-errs: - t.Error(err) - } - vh.Stop() + return true + }) + h.Start() + h.SendEvery(100*time.Millisecond, "a", "10.0.0.1", "10.0.0.3", []byte{222}, 1) + await(h, expired, 10*time.Second, "packet with expired TTL") + }) }