fix: two fixes for dhcp/ap (#136)

* feature: ipv4 broadcast support

* stackip4 method receiver varname

* Incorporate strict test timing for TestStackGoTCPDialRetriesPendingControl (#135)

* rewrite tinygo failing test to be more real-time

* use channels to scheduler stack

* fix flakiness by increasing timeout and move implementation to top of file

* fix(dhcp): use chaddr as lookup key and patch Ethernet dst on Offer/Ack

- Client lookup in Demux() now keys on chaddr when no OptClientIdentifier
  is present. It previously used ciaddr which is 0.0.0.0 during initial lease
  acquisition, causing MsgRequest to fail to match any client (RFC 2131 §4.3.1)
- In Encapsulate(), overwrite Ethernet dst with client.hwaddr when packet is
  embedded in an IP frame (offsetToIP >= 14), per RFC 2131 §4.1. It previously
  sent to gwmac which clients without an ARP entry could not receive

Signed-off-by: deadprogram <ron@hybridgroup.com>

* fix: accept 255.255.255.255 aka bradcast dst in demux4

This fixes a problem with accept 255.255.255.255 dst in demux4 which previously dropped when stack had a static IP.

Signed-off-by: deadprogram <ron@hybridgroup.com>

---------

Signed-off-by: deadprogram <ron@hybridgroup.com>
Co-authored-by: Patricio Whittingslow <graded.sp@gmail.com>

.github/workflows/ci.yaml

@@ -20,9 +20,9 @@ jobs:
 
       - name: gofmt
         run: |
-          unformatted=$(gofmt -l .)
+          unformatted=$(gofmt -l -d . 2>&1) || true
           if [ -n "$unformatted" ]; then
-            echo "::error::Files not formatted with gofmt:"
+            echo "::error::File(s) not formatted with gofmt:"
             echo "$unformatted"
             exit 1
           fi

README.md

@@ -49,6 +49,7 @@ You may try lneto out on linux with the [xcurl example](./examples/xcurl/) which
 - HTTP over TCP/IPv4/Ethernet connection using
 - NTP time check (optional)
 - Print packet captures using lneto's [internet/pcap](./internet/pcap) package
+- Optional fixed TCP source port with `-port`, useful when debugging raw-socket interactions with the host OS
 
 See Developing section below for more information.
 
@@ -171,6 +172,19 @@ sudo ethtool -K <network-interface> gro off
 
 Depending on the interface and driver, other offloading features may also need to be disabled.
 
+When running `xcurl` directly on a normal Linux interface (for example `-i wlan0` or `-i eth0`), lneto shares the host interface IP and MAC address. The Linux TCP stack also sees incoming packets for that IP. If xcurl's TCP handshake succeeds but the HTTP request is followed by a TCP RST from the server, the host kernel may have sent its own outbound RST for xcurl's raw TCP flow because no kernel socket owns that source port.
+
+For a short validation test, use a fixed xcurl source port and temporarily drop kernel-generated outbound RSTs for that port:
+
+```sh
+go build -o examples/xcurl/xcurl ./examples/xcurl
+sudo iptables -I OUTPUT -p tcp --sport 2300 --tcp-flags RST RST -j DROP
+sudo ./examples/xcurl/xcurl -i <network-interface> -port 2300 -host example.com
+sudo iptables -D OUTPUT -p tcp --sport 2300 --tcp-flags RST RST -j DROP
+```
+
+This is a debugging workaround, not the preferred operating mode. Prefer `-ihttp`, a TAP interface, a network namespace, or an otherwise isolated interface/IP when testing xcurl without host TCP stack interference.
+
 - [`examples/httptap`](./examples/httptap) (linux only, root privilidges required) Program opens a TAP interface and assigns an IP address to it and exposes the interface via a HTTP interface. This program is run with root privilidges to facilitate debugging of lneto since no root privilidges are required to interact with the HTTP interface exposed.
     - `POST http://127.0.0.1:7070/send`: Receives a POST with request body containing JSON string of data to send over TAP interface. Response contains only status code.
     - `GET http://127.0.0.1:7070/recv`: Receives a GET request. Response contains a JSON string of oldest unread TAP interface packet. If string is empty then there is no more data to read.

dhcp/dhcpv4/server.go

@@ -152,7 +152,20 @@ func (sv *Server) Demux(carrierData []byte, frameOffset int) error {
 	var client serverEntry
 	var clientExists bool
 	if len(clientID) == 0 {
-		client, clientIDRaw, clientExists = sv.getClientByIP(*dfrm.CIAddr())
+		// No explicit client identifier: use chaddr as the stable lookup key.
+		// This lets the server correlate Discover→Offer→Request even when
+		// ciaddr=0.0.0.0 (client has no IP yet), which is the normal case
+		// for first-time lease acquisition (RFC 2131 §4.3.1).
+		chaddr := *dfrm.CHAddrAs6()
+		copy(clientIDRaw[:], chaddr[:])
+		client, clientExists = sv.getClient(clientIDRaw)
+		if !clientExists {
+			// Fallback: look up by ciaddr for clients that did send ciaddr.
+			ciaddr := *dfrm.CIAddr()
+			if ciaddr != ([4]byte{}) {
+				client, clientIDRaw, clientExists = sv.getClientByIP(ciaddr)
+			}
+		}
 	} else {
 		copy(clientIDRaw[:], clientID)
 		client, clientExists = sv.getClient(clientIDRaw)
@@ -301,6 +314,14 @@ func (sv *Server) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int)
 		if err != nil {
 			return 0, err
 		}
+		// Per RFC 2131 §4.1: unicast Offer/Ack to chaddr because the client
+		// does not yet have the offered IP, so no ARP entry exists.
+		// The Ethernet layer sets dst=gwmac before calling us; overwrite it
+		// here so the frame reaches the client via its hardware address.
+		// offsetToIP==14 means the Ethernet header is at carrierData[0:14].
+		if offsetToIP >= 14 {
+			copy(carrierData[offsetToIP-14:offsetToIP-8], client.hwaddr[:])
+		}
 	}
 
 	client.state = futureState

dhcp/dhcpv4/server_ap_test.go

@@ -0,0 +1,252 @@
+package dhcpv4
+
+// Tests covering the AP-mode DHCP server fixes:
+//   1. Client lookup by chaddr when no ClientID option is present.
+//   2. Ciaddr fallback lookup when chaddr lookup finds nothing.
+//   3. Ethernet dst is patched to chaddr on Offer/Ack when Encapsulate is
+//      called with offsetToIP >= 14.
+
+import (
+	"testing"
+
+	"github.com/soypat/lneto/ipv4"
+)
+
+// doraNoClientID runs a complete DORA exchange using raw DHCP frames
+// (no ClientID option) with the given chaddr. The client uses only chaddr
+// as its identity, matching normal AP-mode behaviour where mobile clients
+// don't send OptClientIdentifier.
+func doraNoClientID(t *testing.T, sv *Server, chaddr [6]byte, xid uint32) (assignedIP [4]byte) {
+	t.Helper()
+	var cl Client
+	err := cl.BeginRequest(xid, RequestConfig{
+		ClientHardwareAddr: chaddr,
+		// Intentionally no ClientID – forces server to key on chaddr.
+	})
+	if err != nil {
+		t.Fatalf("BeginRequest: %v", err)
+	}
+
+	var buf [1024]byte
+
+	// DISCOVER
+	n, err := cl.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("discover encapsulate: n=%d err=%v", n, err)
+	}
+	if err = sv.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("discover demux: %v", err)
+	}
+
+	// OFFER
+	n, err = sv.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("offer encapsulate: n=%d err=%v", n, err)
+	}
+	frm, _ := NewFrame(buf[:n])
+	assignedIP = *frm.YIAddr()
+	if err = cl.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("offer demux: %v", err)
+	}
+
+	// REQUEST
+	n, err = cl.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("request encapsulate: n=%d err=%v", n, err)
+	}
+	if err = sv.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("request demux: %v", err)
+	}
+
+	// ACK
+	n, err = sv.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("ack encapsulate: n=%d err=%v", n, err)
+	}
+	if err = cl.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("ack demux: %v", err)
+	}
+
+	if cl.State() != StateBound {
+		t.Errorf("want StateBound, got %s", cl.State())
+	}
+	return assignedIP
+}
+
+// TestServerChaddrLookup_NoClientID verifies the server can complete a full DORA
+// cycle when the client sends no OptClientIdentifier option.  Before the fix the
+// MsgRequest Demux call would fail with "request for non existing client" because
+// the server tried to look up the entry by ciaddr (0.0.0.0) instead of chaddr.
+func TestServerChaddrLookup_NoClientID(t *testing.T) {
+	svAddr := [4]byte{192, 168, 4, 1}
+	var sv Server
+	sv.Configure(ServerConfig{
+		ServerAddr: svAddr,
+		Gateway:    svAddr,
+		Subnet:     ipv4.PrefixFrom(svAddr, 24),
+	})
+
+	chaddr := [6]byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}
+	ip := doraNoClientID(t, &sv, chaddr, 0x12345678)
+	if ip[0] != 192 || ip[1] != 168 || ip[2] != 4 {
+		t.Errorf("unexpected subnet in assigned IP %v", ip)
+	}
+}
+
+// TestServerChaddrLookup_MultipleClientsNoClientID verifies that multiple clients
+// without ClientID each get a unique address and successfully reach StateBound.
+func TestServerChaddrLookup_MultipleClientsNoClientID(t *testing.T) {
+	svAddr := [4]byte{192, 168, 4, 1}
+	var sv Server
+	sv.Configure(ServerConfig{
+		ServerAddr: svAddr,
+		Gateway:    svAddr,
+		Subnet:     ipv4.PrefixFrom(svAddr, 24),
+	})
+
+	addrs := make([][4]byte, 4)
+	for i := range addrs {
+		chaddr := [6]byte{0x00, 0x11, 0x22, 0x33, 0x44, byte(i + 1)}
+		addrs[i] = doraNoClientID(t, &sv, chaddr, uint32(i+1))
+	}
+
+	for i := range addrs {
+		for j := i + 1; j < len(addrs); j++ {
+			if addrs[i] == addrs[j] {
+				t.Errorf("clients %d and %d got same address %v", i, j, addrs[i])
+			}
+		}
+	}
+}
+
+// TestServerCiaddrFallback exercises the ciaddr fallback path in Demux.
+// An entry is registered under an explicit non-MAC ClientID.  A subsequent
+// RELEASE arrives with no ClientID (so chaddr lookup misses) but with
+// ciaddr=assignedIP, which should let the server find and remove the entry.
+func TestServerCiaddrFallback(t *testing.T) {
+	svAddr := [4]byte{192, 168, 4, 1}
+	chaddr := [6]byte{0xca, 0xfe, 0x00, 0x00, 0x01, 0x01}
+	const xid = uint32(0xaabbccdd)
+	var sv Server
+	sv.Configure(ServerConfig{
+		ServerAddr: svAddr,
+		Subnet:     ipv4.PrefixFrom(svAddr, 24),
+	})
+
+	// DORA with an explicit ClientID — server keys the entry by that string.
+	var cl Client
+	if err := cl.BeginRequest(xid, RequestConfig{
+		ClientHardwareAddr: chaddr,
+		ClientID:           "device-id",
+	}); err != nil {
+		t.Fatalf("BeginRequest: %v", err)
+	}
+	assignedIP := doraWithClient(t, &sv, &cl)
+
+	// RELEASE: no ClientID, ciaddr = assignedIP.
+	// chaddr lookup fails (entry is keyed by "device-id"), so server must
+	// fall back to getClientByIP.
+	var relBuf [512]byte
+	rfrm, _ := NewFrame(relBuf[:])
+	rfrm.ClearHeader()
+	rfrm.SetOp(OpRequest)
+	rfrm.SetHardware(1, 6, 0)
+	rfrm.SetXID(xid)
+	*rfrm.CIAddr() = assignedIP
+	copy(rfrm.CHAddrAs6()[:], chaddr[:])
+	rfrm.SetMagicCookie(MagicCookie)
+	opts := rfrm.OptionsPayload()
+	n, _ := EncodeOption(opts, OptMessageType, byte(MsgRelease))
+	opts[n] = byte(OptEnd)
+	n++
+	if err := sv.Demux(relBuf[:OptionsOffset+n], 0); err != nil {
+		t.Fatalf("release demux: %v", err)
+	}
+	if len(sv.hosts) != 0 {
+		t.Errorf("expected 0 hosts after release, got %d", len(sv.hosts))
+	}
+}
+
+// TestServerOfferPatchesEthernetDst verifies that Encapsulate overwrites the
+// first 6 bytes of the carrier (Ethernet dst) with the client's chaddr when
+// offsetToIP >= 14.
+func TestServerOfferPatchesEthernetDst(t *testing.T) {
+	svAddr := [4]byte{192, 168, 4, 1}
+	clChaddr := [6]byte{0x02, 0x03, 0x04, 0x05, 0x06, 0x07}
+	var sv Server
+	sv.Configure(ServerConfig{
+		ServerAddr: svAddr,
+		Subnet:     ipv4.PrefixFrom(svAddr, 24),
+	})
+
+	var cl Client
+	cl.BeginRequest(0x11223344, RequestConfig{ClientHardwareAddr: clChaddr})
+
+	var buf [1024]byte
+	n, err := cl.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("discover encapsulate: n=%d err=%v", n, err)
+	}
+	if err = sv.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("discover demux: %v", err)
+	}
+
+	// Carrier: 14 bytes Ethernet header, then a minimal IPv4 header marker.
+	var carrier [1024]byte
+	carrier[14] = 0x45 // IPv4 version+IHL so SetIPAddrs succeeds.
+	offerN, err := sv.Encapsulate(carrier[:], 14, 14+20+8)
+	if err != nil || offerN == 0 {
+		t.Fatalf("offer encapsulate: n=%d err=%v", offerN, err)
+	}
+
+	var got [6]byte
+	copy(got[:], carrier[0:6])
+	if got != clChaddr {
+		t.Errorf("Ethernet dst: got %v, want %v", got, clChaddr)
+	}
+}
+
+// doraWithClient runs a complete DORA exchange using the given pre-configured
+// Client and returns the assigned IP.
+func doraWithClient(t *testing.T, sv *Server, cl *Client) (assignedIP [4]byte) {
+	t.Helper()
+	var buf [1024]byte
+
+	n, err := cl.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("discover encapsulate: n=%d err=%v", n, err)
+	}
+	if err = sv.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("discover demux: %v", err)
+	}
+
+	n, err = sv.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("offer encapsulate: n=%d err=%v", n, err)
+	}
+	frm, _ := NewFrame(buf[:n])
+	assignedIP = *frm.YIAddr()
+	if err = cl.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("offer demux: %v", err)
+	}
+
+	n, err = cl.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("request encapsulate: n=%d err=%v", n, err)
+	}
+	if err = sv.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("request demux: %v", err)
+	}
+
+	n, err = sv.Encapsulate(buf[:], -1, 0)
+	if err != nil || n == 0 {
+		t.Fatalf("ack encapsulate: n=%d err=%v", n, err)
+	}
+	if err = cl.Demux(buf[:n], 0); err != nil {
+		t.Fatalf("ack demux: %v", err)
+	}
+	if cl.State() != StateBound {
+		t.Errorf("want StateBound, got %s", cl.State())
+	}
+	return assignedIP
+}

examples/xcurl/main.go

@@ -50,6 +50,7 @@ func run() (err error) {
 		flagUseHTTP       = false
 		flagHostToResolve = ""
 		flagRequestedIP   = ""
+		flagLocalTCPPort  = 0
 		flagDoNTP         = false
 		flagNoPcap        = false
 		flagPprof         = false
@@ -58,6 +59,7 @@ func run() (err error) {
 	flag.BoolVar(&flagUseHTTP, "ihttp", flagUseHTTP, "Use HTTP tap interface.")
 	flag.StringVar(&flagHostToResolve, "host", flagHostToResolve, "Hostname to resolve via DNS.")
 	flag.StringVar(&flagRequestedIP, "addr", flagRequestedIP, "IP address to request via DHCP.")
+	flag.IntVar(&flagLocalTCPPort, "port", flagLocalTCPPort, "Local TCP source port to use. Zero chooses a pseudo-random port.")
 	flag.BoolVar(&flagDoNTP, "ntp", flagDoNTP, "Do NTP round and print result time")
 	flag.BoolVar(&flagNoPcap, "nopcap", flagNoPcap, "Disable pcap logging.")
 	flag.BoolVar(&flagPprof, "pprof", flagPprof, "Enable CPU profiling.")
@@ -79,6 +81,21 @@ func run() (err error) {
 		flag.Usage()
 		return err
 	}
+	reqAddr := [4]byte{192, 168, 1, 96}
+	requestedAddrSet := false
+	if flagRequestedIP != "" {
+		addr, err := netip.ParseAddr(flagRequestedIP)
+		if err != nil || !addr.Is4() {
+			flag.Usage()
+			return fmt.Errorf("invalid requested IPv4 address %q", flagRequestedIP)
+		}
+		reqAddr = addr.As4()
+		requestedAddrSet = true
+	}
+	if flagLocalTCPPort < 0 || flagLocalTCPPort > math.MaxUint16 {
+		flag.Usage()
+		return fmt.Errorf("invalid local TCP port %d", flagLocalTCPPort)
+	}
 	fmt.Println("softrand", softRand)
 	var iface ltesto.Interface
 	if flagUseHTTP {
@@ -223,11 +240,14 @@ func run() (err error) {
 		dhcpRetries = 2
 	)
 	timeDHCP := timer("DHCP request completed")
-	results, err := rstack.DoDHCPv4([4]byte{192, 168, 1, 96}, dhcpTimeout, dhcpRetries)
+	results, err := rstack.DoDHCPv4(reqAddr, dhcpTimeout, dhcpRetries)
 	if err != nil {
 		return fmt.Errorf("DHCP failed: %w", err)
 	}
 	timeDHCP()
+	if requestedAddrSet && results.AssignedAddr4 != reqAddr {
+		return fmt.Errorf("DHCP assigned %s, not requested %s", netip.AddrFrom4(results.AssignedAddr4), netip.AddrFrom4(reqAddr))
+	}
 
 	err = stack.AssimilateDHCPResults(results)
 	if err != nil {
@@ -302,7 +322,12 @@ func run() (err error) {
 	timeTCPDial := timer("TCP dial (handshake)")
 	const tcpDialTimeout = 8 * time.Second // Was 60 * time.Minute causing 3.6s sleep per iteration!
 	target := netip.AddrPortFrom(addrs[0], 80)
-	err = rstack.DoDialTCP(&conn, uint16(softRand&0xefff)+1024, target, tcpDialTimeout, internetRetries)
+	localPort := uint16(flagLocalTCPPort)
+	if localPort == 0 {
+		localPort = uint16(softRand&0xefff) + 1024
+	}
+	fmt.Printf("TCP target %s local-port=%d\n", target, localPort)
+	err = rstack.DoDialTCP(&conn, localPort, target, tcpDialTimeout, internetRetries)
 	if err != nil {
 		return fmt.Errorf("TCP failed: %w", err)
 	}

internet/conn_closerace_test.go

@@ -0,0 +1,214 @@
+package internet
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"math/rand"
+	"runtime"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/soypat/lneto/tcp"
+)
+
+// TestConn_ConcurrentCloseDoesNotDropWrite is a regression test for issue #82:
+// when one goroutine has a Write in progress (blocked because the TX buffer is
+// full) and another goroutine calls Close, the in-flight write data must not be
+// silently dropped. With the atomic write-lock, Close waits for the in-progress
+// write to finish queueing all of its data before tearing the connection down.
+//
+// The scenario is made deterministic by filling the TX buffer before any packets
+// are exchanged: the writer blocks holding the write lock, Close is then issued
+// (and must wait), and only afterwards is the packet pump started so the buffer
+// can drain and the writer can run to completion.
+func TestConn_ConcurrentCloseDoesNotDropWrite(t *testing.T) {
+	rng := rand.New(rand.NewSource(1))
+	var sbCl, sbSv StackIPv4
+	var connCl, connSv tcp.Conn
+	setupClientServerEstablished(t, rng, &sbCl, &sbSv, &connCl, &connSv)
+
+	// Payload several times larger than the 2048-byte TX buffer so the writer
+	// must block waiting for the buffer to drain across many segments.
+	payload := make([]byte, 4*2048)
+	for i := range payload {
+		payload[i] = byte(i)
+	}
+
+	// Watchdog: break any deadlock so a regression fails fast instead of hanging.
+	watchdog := time.AfterFunc(10*time.Second, func() {
+		t.Error("test timed out: Close likely blocked waiting on a stuck write")
+		connCl.Abort()
+		connSv.Abort()
+	})
+	defer watchdog.Stop()
+
+	// Writer (G1): writes the whole payload. Blocks once the TX buffer fills.
+	type writeResult struct {
+		n   int
+		err error
+	}
+	writeDone := make(chan writeResult, 1)
+	go func() {
+		n, err := connCl.Write(payload)
+		writeDone <- writeResult{n, err}
+	}()
+
+	// Wait until the writer has filled the TX buffer and is blocked. At this
+	// point it owns the write lock, reproducing "Write in progress".
+	for connCl.FreeOutput() != 0 {
+		runtime.Gosched()
+	}
+
+	// Close (G2): issued while the write is in progress. Must wait for the
+	// writer to drain instead of dropping its data.
+	closeDone := make(chan error, 1)
+	go func() {
+		closeDone <- connCl.Close()
+	}()
+
+	// Reader: drains the server side until EOF, accumulating everything received.
+	readDone := make(chan []byte, 1)
+	go func() {
+		got := make([]byte, 0, len(payload))
+		rbuf := make([]byte, 1024)
+		for {
+			n, err := connSv.Read(rbuf)
+			got = append(got, rbuf[:n]...)
+			if err != nil {
+				break
+			}
+		}
+		readDone <- got
+	}()
+
+	// Packet pump: only now do we move packets between the two stacks, letting
+	// the buffer drain so the blocked writer can finish.
+	var stop atomic.Bool
+	pumpStopped := make(chan struct{})
+	go func() {
+		defer close(pumpStopped)
+		var buf [2048]byte
+		for !stop.Load() {
+			progressed := false
+			if n, err := sbCl.Encapsulate(buf[:], 0, 0); err == nil && n > 0 {
+				_ = sbSv.Demux(buf[:n], 0)
+				progressed = true
+			}
+			if n, err := sbSv.Encapsulate(buf[:], 0, 0); err == nil && n > 0 {
+				_ = sbCl.Demux(buf[:n], 0)
+				progressed = true
+			}
+			if !progressed {
+				runtime.Gosched()
+			}
+		}
+	}()
+
+	wr := <-writeDone
+	if wr.err != nil {
+		t.Errorf("issue #82: Write returned error %v; concurrent Close dropped in-flight data", wr.err)
+	}
+	if wr.n != len(payload) {
+		t.Errorf("issue #82: Write wrote %d of %d bytes; concurrent Close truncated the write", wr.n, len(payload))
+	}
+	if err := <-closeDone; err != nil {
+		t.Errorf("Close returned error: %v", err)
+	}
+	got := <-readDone
+	stop.Store(true)
+	<-pumpStopped
+
+	if len(got) != len(payload) {
+		t.Fatalf("server received %d of %d bytes; data was dropped by concurrent Close", len(got), len(payload))
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatal("server received corrupted data")
+	}
+}
+
+// TestConn_ConcurrentWritesSerialize verifies that the atomic write-lock
+// serializes concurrent Write calls on the same connection so their payloads are
+// not interleaved on the wire, and that all bytes from both writers are delivered.
+func TestConn_ConcurrentWritesSerialize(t *testing.T) {
+	rng := rand.New(rand.NewSource(2))
+	var sbCl, sbSv StackIPv4
+	var connCl, connSv tcp.Conn
+	setupClientServerEstablished(t, rng, &sbCl, &sbSv, &connCl, &connSv)
+
+	const chunk = 1500
+	a := bytes.Repeat([]byte{0xAA}, chunk)
+	b := bytes.Repeat([]byte{0xBB}, chunk)
+
+	watchdog := time.AfterFunc(10*time.Second, func() {
+		t.Error("test timed out")
+		connCl.Abort()
+		connSv.Abort()
+	})
+	defer watchdog.Stop()
+
+	var stop atomic.Bool
+	pumpStopped := make(chan struct{})
+	go func() {
+		defer close(pumpStopped)
+		var buf [2048]byte
+		for !stop.Load() {
+			progressed := false
+			if n, err := sbCl.Encapsulate(buf[:], 0, 0); err == nil && n > 0 {
+				_ = sbSv.Demux(buf[:n], 0)
+				progressed = true
+			}
+			if n, err := sbSv.Encapsulate(buf[:], 0, 0); err == nil && n > 0 {
+				_ = sbCl.Demux(buf[:n], 0)
+				progressed = true
+			}
+			if !progressed {
+				runtime.Gosched()
+			}
+		}
+	}()
+
+	writeErr := make(chan error, 2)
+	writer := func(b []byte) {
+		n, err := connCl.Write(b)
+		if err == nil && n != len(b) {
+			err = io.ErrShortWrite
+		}
+		writeErr <- err
+	}
+	go writer(a)
+	go writer(b)
+
+	got := make([]byte, 0, 2*chunk)
+	rbuf := make([]byte, 1024)
+	for len(got) < 2*chunk {
+		n, err := connSv.Read(rbuf)
+		got = append(got, rbuf[:n]...)
+		if err != nil && !errors.Is(err, io.EOF) {
+			break
+		}
+	}
+	for range 2 {
+		if err := <-writeErr; err != nil {
+			t.Errorf("concurrent write failed: %v", err)
+		}
+	}
+	stop.Store(true)
+	<-pumpStopped
+
+	if len(got) != 2*chunk {
+		t.Fatalf("received %d bytes, want %d", len(got), 2*chunk)
+	}
+	// Serialized writes mean one chunk's bytes appear fully before the other's;
+	// the boundary is a single transition, never interleaved.
+	transitions := 0
+	for i := 1; i < len(got); i++ {
+		if got[i] != got[i-1] {
+			transitions++
+		}
+	}
+	if transitions != 1 {
+		t.Fatalf("expected exactly one A/B boundary (serialized writes), got %d transitions", transitions)
+	}
+}

internet/stack-ip4.go

@@ -22,34 +22,35 @@ type StackIPv4 struct {
 }
 
 func (stackip4 *StackIPv4) Reset(vld *lneto.Validator, maxNodes int) error {
+	stackip4.connID++
 	stackip4.reset4(vld, maxNodes)
 	return nil
 }
 
-func (stackip *StackIPv4) ConnectionID() *uint64 {
-	return &stackip.connID
+func (stackip4 *StackIPv4) ConnectionID() *uint64 {
+	return &stackip4.connID
 }
 
-func (stackip *StackIPv4) Protocol() uint64 {
+func (stackip4 *StackIPv4) Protocol() uint64 {
 	return uint64(ethernet.TypeIPv4)
 }
 
-func (stackip *StackIPv4) LocalPort() uint16 { return 0 }
+func (stackip4 *StackIPv4) LocalPort() uint16 { return 0 }
 
-func (stackip *StackIPv4) SetLogger(logger *slog.Logger) {
-	stackip.stackip4.handlers.log = logger
+func (stackip4 *StackIPv4) SetLogger(logger *slog.Logger) {
+	stackip4.stackip4.handlers.log = logger
 }
 
-func (stackip *StackIPv4) Demux(carrierData []byte, offset int) error {
+func (stackip4 *StackIPv4) Demux(carrierData []byte, offset int) error {
 	debugLog("ip:demux")
-	return stackip.stackip4.demux4(carrierData, offset)
+	return stackip4.stackip4.demux4(carrierData, offset)
 }
 
-func (stackip *StackIPv4) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (n int, err error) {
+func (stackip4 *StackIPv4) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (n int, err error) {
 	if offsetToFrame != offsetToIP {
 		return 0, lneto.ErrBug
 	}
-	return stackip.stackip4.encapsulate4(carrierData, offsetToIP)
+	return stackip4.stackip4.encapsulate4(carrierData, offsetToIP)
 }
 
 type stackip4 struct {
@@ -58,6 +59,7 @@ type stackip4 struct {
 	ipID            uint16
 	ip4             [4]byte
 	acceptMulticast bool
+	acceptBroadcast bool
 }
 
 func (si4 *stackip4) reset4(vld *lneto.Validator, maxNodes int) {
@@ -86,6 +88,10 @@ func (si4 *stackip4) IsRegistered4(proto lneto.IPProto) bool {
 func (si4 *stackip4) SetAcceptMulticast4(accept bool) {
 	si4.acceptMulticast = accept
 }
+func (si4 *stackip4) SetAcceptBroadcast4(accept bool) {
+	si4.acceptBroadcast = accept
+}
+
 func (si4 *stackip4) Addr4() [4]byte { return si4.ip4 }
 func (si4 *stackip4) SetAddr4(ip4 [4]byte) {
 	si4.ip4 = ip4
@@ -100,8 +106,13 @@ func (si4 *stackip4) demux4(carrierData []byte, offset int) error {
 		return err
 	}
 	dst := ifrm.DestinationAddr()
-	if si4.ip4 != ([4]byte{}) && *dst != si4.ip4 {
-		if !si4.acceptMulticast || !internal.IsMulticastIPAddr(dst[:]) {
+	if si4.ip4 != ([4]byte{}) && *dst != si4.ip4 { // Accept all packets when IP zeroed.
+		switch {
+		case si4.acceptMulticast && ipv4.IsMulticast(*dst):
+			// accept multicast.
+		case si4.acceptBroadcast && ipv4.IsBroadcast(*dst):
+			// accept broadcast.
+		default:
 			si4.handlers.debug("ip:not-for-us")
 			return lneto.ErrPacketDrop // Not meant for us.
 		}

internet/stack-ip6.go

@@ -19,35 +19,36 @@ type StackIPv6 struct {
 	stackip6
 }
 
-func (stackip4 *StackIPv6) Reset(vld *lneto.Validator, maxNodes int) error {
-	stackip4.reset6(vld, maxNodes)
+func (stackip6 *StackIPv6) Reset(vld *lneto.Validator, maxNodes int) error {
+	stackip6.connID++
+	stackip6.reset6(vld, maxNodes)
 	return nil
 }
 
-func (stackip *StackIPv6) ConnectionID() *uint64 {
-	return &stackip.connID
+func (stackip6 *StackIPv6) ConnectionID() *uint64 {
+	return &stackip6.connID
 }
 
-func (stackip *StackIPv6) Protocol() uint64 {
+func (stackip6 *StackIPv6) Protocol() uint64 {
 	return uint64(ethernet.TypeIPv6)
 }
 
-func (stackip *StackIPv6) LocalPort() uint16 { return 0 }
+func (stackip6 *StackIPv6) LocalPort() uint16 { return 0 }
 
-func (stackip *StackIPv6) SetLogger(logger *slog.Logger) {
-	stackip.stackip6.handlers.log = logger
+func (stackip6 *StackIPv6) SetLogger(logger *slog.Logger) {
+	stackip6.stackip6.handlers.log = logger
 }
 
-func (stackip *StackIPv6) Demux(carrierData []byte, offset int) error {
+func (stackip6 *StackIPv6) Demux(carrierData []byte, offset int) error {
 	debugLog("ip:demux")
-	return stackip.stackip6.demux6(carrierData, offset)
+	return stackip6.stackip6.demux6(carrierData, offset)
 }
 
-func (stackip *StackIPv6) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (n int, err error) {
+func (stackip6 *StackIPv6) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (n int, err error) {
 	if offsetToFrame != offsetToIP {
 		return 0, lneto.ErrBug
 	}
-	return stackip.stackip6.encapsulate6(carrierData, offsetToIP)
+	return stackip6.stackip6.encapsulate6(carrierData, offsetToIP)
 }
 
 type stackip6 struct {
@@ -55,6 +56,7 @@ type stackip6 struct {
 	vld             *lneto.Validator
 	ip6             [16]byte
 	acceptMulticast bool
+	acceptBroadcast bool
 }
 
 func (si6 *stackip6) Register6(h lneto.StackNode) error {

ipv4/definitions.go

@@ -18,6 +18,17 @@ func IsMulticast(addr [4]byte) bool {
 	return addr[0]&0xf0 == 0xe0
 }
 
+// IsBroadcast reports whether addr is the limited broadcast address
+// 255.255.255.255 used to address all hosts on the local network segment
+// as defined in [RFC919]. It does not detect directed (subnet) broadcast
+// addresses, which depend on the network mask and cannot be determined from
+// the address alone.
+//
+// [RFC919]: https://datatracker.ietf.org/doc/html/rfc919
+func IsBroadcast(addr [4]byte) bool {
+	return addr == [4]byte{255, 255, 255, 255}
+}
+
 // IsLinkLocal reports whether addr is within the IPv4 link-local prefix
 // 169.254.0.0/16 reserved for dynamic link-local configuration as defined in [RFC3927].
 //

tcp/conn.go

@@ -8,6 +8,7 @@ import (
 	"net/netip"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/soypat/lneto"
@@ -29,6 +30,13 @@ type Conn struct {
 	h          Handler
 	remoteAddr []byte
 
+	// writeLock holds the connection ID ([Handler.connid]) of the goroutine
+	// that currently owns the write side via [Conn.Write] or [Conn.Flush].
+	// A zero value means no write is in progress. It serializes concurrent
+	// writers and lets [Conn.Close] acquire the write side before closing so it
+	// does not drop in-flight data (see issue #82).
+	writeLock atomic.Uint64
+
 	_backoff lneto.BackoffStrategy
 	rdead    time.Time
 	wdead    time.Time
@@ -49,6 +57,7 @@ func (conn *Conn) reset(h Handler) {
 	conn.wdead = time.Time{}
 	conn.abortErr = nil
 	conn.ipID = 0
+	conn.writeLock.Store(0)
 }
 
 // ConnConfig provides configuration parameters for [Conn].
@@ -99,6 +108,22 @@ func (conn *Conn) RemotePort() uint16 {
 	return conn.h.RemotePort()
 }
 
+// IsAwaitingControl reports whether the connection is waiting for a response to
+// a control segment that can be retransmitted to advance connection state.
+func (conn *Conn) IsAwaitingControl() bool {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	return conn.h.IsAwaitingControl()
+}
+
+// RequeueControl asks the next packet emission to retransmit the outstanding
+// control segment, if the connection is waiting for one.
+func (conn *Conn) RequeueControl() {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	conn.h.RequeueControl()
+}
+
 // RemoteAddr returns the address of the peer Conn is exchanging data with.
 func (conn *Conn) RemoteAddr() []byte {
 	conn.mu.Lock()
@@ -213,6 +238,11 @@ func (conn *Conn) CloseRead() error {
 
 // Close will initiate TCP close sequence. After Close is called future [Conn.Write] calls will fail with [net.ErrClosed].
 func (conn *Conn) Close() error {
+	connid, err := conn.acquireWriteLock(nil)
+	if err != nil {
+		return err
+	}
+	defer conn.releaseWriteLock(connid)
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 	conn.trace("TCPConn.Close", slog.Uint64("lport", uint64(conn.h.localPort)), slog.Uint64("rport", uint64(conn.h.remotePort)))
@@ -237,10 +267,11 @@ func (conn *Conn) InternalHandler() *Handler {
 
 // Write writes argument data to the TCPConns's output buffer which is queued to be sent.
 func (conn *Conn) Write(b []byte) (int, error) {
-	connid, err := conn.lockPipeConnID()
+	connid, err := conn.acquireWriteLock(&conn.wdead)
 	if err != nil {
 		return 0, err
 	}
+	defer conn.releaseWriteLock(connid)
 	rport := conn.RemotePort()
 	plen := len(b)
 	lport := conn.LocalPort()
@@ -286,10 +317,11 @@ func (conn *Conn) Write(b []byte) (int, error) {
 
 // Flush blocks until all buffered TCP data has been sent.
 func (conn *Conn) Flush() error {
-	connid, err := conn.lockPipeConnID()
+	connid, err := conn.acquireWriteLock(&conn.wdead)
 	if err != nil {
 		return err
 	}
+	defer conn.releaseWriteLock(connid)
 	var backoffs uint
 	for conn.BufferedUnsent() != 0 {
 		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
@@ -350,14 +382,31 @@ func (conn *Conn) Read(b []byte) (int, error) {
 	return n, nil
 }
 
-func (conn *Conn) lockPipeConnID() (uint64, error) {
+// acquireWriteLock validates the connection is open and acquires the write lock
+// for the current connection, returning its connection ID. It serializes
+// concurrent writers: while another goroutine owns the write side it blocks with
+// backoff until that writer releases, the connection closes, or deadline elapses.
+// The returned connID must be released with [Conn.releaseWriteLock].
+func (conn *Conn) acquireWriteLock(deadline *time.Time) (connID uint64, err error) {
 	conn.mu.Lock()
-	defer conn.mu.Unlock()
-	err := conn.checkPipeOpen()
-	if err != nil {
-		return 0, err
+	connID = conn.h.connid
+	conn.mu.Unlock()
+	var backoffs uint
+	for {
+		if err := conn.checkPipe(connID, deadline); err != nil {
+			return 0, err
+		}
+		if conn.writeLock.CompareAndSwap(0, connID) {
+			return connID, nil
+		}
+		conn.backoff(backoffs)
+		backoffs++
 	}
-	return conn.h.connid, nil
+}
+
+// releaseWriteLock releases a write lock previously acquired by [Conn.acquireWriteLock].
+func (conn *Conn) releaseWriteLock(connID uint64) {
+	conn.writeLock.CompareAndSwap(connID, 0)
 }
 
 func (conn *Conn) checkPipe(connID uint64, deadline *time.Time) (err error) {
@@ -365,9 +414,9 @@ func (conn *Conn) checkPipe(connID uint64, deadline *time.Time) (err error) {
 	defer conn.mu.Unlock()
 	if conn.abortErr != nil {
 		err = conn.abortErr
-	} else if connID != conn.h.connid {
+	} else if connID != conn.h.connid || conn.h.State().IsClosed() {
 		err = net.ErrClosed
-	} else if !deadline.IsZero() && time.Since(*deadline) > 0 {
+	} else if deadline != nil && !deadline.IsZero() && time.Since(*deadline) > 0 {
 		err = errDeadlineExceeded
 	}
 	return err

tcp/handler.go

@@ -32,7 +32,8 @@ type Handler struct {
 	closing    bool
 	shutdownRx bool
 	// nRetransmit stores the number of times the oldest packet was retransmit.
-	nRetransmit uint8
+	nRetransmit    uint8
+	requeueControl bool
 }
 
 // SetLoggers sets the [slog.Logger] for the Handler and internal [ControlBlock].
@@ -271,6 +272,7 @@ func (h *Handler) Send(b []byte) (int, error) {
 		return 0, net.ErrClosed
 	}
 	awaitingSyn := h.AwaitingSynSend()
+	requeueControl := h.requeueControl
 	buffered := h.bufTx.BufferedUnsent()
 	if h.scb.State() == StateCloseWait && !h.closing && buffered == 0 && !h.scb.HasPending() {
 		// Remote closed with no application data left to send: initiate our own close.
@@ -278,7 +280,7 @@ func (h *Handler) Send(b []byte) (int, error) {
 		// before Send is called, implementing the half-close per RFC 9293 §3.5.
 		h.closing = true
 	}
-	if !awaitingSyn && buffered == 0 && !h.closing && !h.scb.HasPending() {
+	if !awaitingSyn && !requeueControl && buffered == 0 && !h.closing && !h.scb.HasPending() {
 		// Early nop short circuit.
 		return 0, nil
 	}
@@ -301,11 +303,27 @@ func (h *Handler) Send(b []byte) (int, error) {
 	offset := uint8(5)
 	mss := uint16(len(b) - sizeHeaderTCP)
 	var segment Segment
-	if awaitingSyn {
+	if awaitingSyn || requeueControl && h.scb.State() == StateSynSent {
 		// Handling init syn segment.
 		segment = ClientSynSegment(h.bufTx.iss, Size(h.bufRx.Size()))
 		h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss)
 		offset++
+		if requeueControl {
+			h.info("tcp.Handler:requeue-syn", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort)))
+		}
+	} else if requeueControl && h.scb.State() == StateSynRcvd {
+		segment = Segment{
+			SEQ:   h.scb.snd.UNA,
+			ACK:   h.scb.rcv.NXT,
+			WND:   Size(h.bufRx.Free()),
+			Flags: synack,
+		}
+		h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss)
+		offset++
+		h.info("tcp.Handler:requeue-synack", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort)))
+	} else if requeueControl {
+		h.requeueControl = false
+		return 0, nil
 	} else {
 		var ok bool
 		maxPayload := len(b) - sizeHeaderTCP
@@ -335,6 +353,7 @@ func (h *Handler) Send(b []byte) (int, error) {
 	} else if prevState != h.scb.State() && h.logenabled(slog.LevelInfo) {
 		h.info("tcp.Handler:tx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("oldState", prevState.String()), slog.String("newState", h.scb.State().String()), slog.String("txflags", segment.Flags.String()))
 	}
+	h.requeueControl = false
 	tfrm.SetSourcePort(h.localPort)
 	tfrm.SetDestinationPort(h.remotePort)
 	tfrm.SetSegment(segment, offset)
@@ -443,6 +462,20 @@ func (h *Handler) AwaitingSynResponse() bool {
 	return h.remotePort != 0 && h.scb.State() == StateSynSent
 }
 
+// IsAwaitingControl reports whether the connection is waiting for a response to
+// a control segment that can be retransmitted to advance connection state.
+func (h *Handler) IsAwaitingControl() bool {
+	return h.AwaitingSynResponse() || h.scb.State() == StateSynRcvd
+}
+
+// RequeueControl asks the next Send call to retransmit the outstanding control
+// segment, if the connection is waiting for one.
+func (h *Handler) RequeueControl() {
+	if h.IsAwaitingControl() {
+		h.requeueControl = true
+	}
+}
+
 // AwaitingSynAck returns true if the Handler is a passive server opened with [Handler.OpenListen] and not yet received a valid SYN remote packet.
 func (h *Handler) AwaitingSynAck() bool {
 	return h.remotePort == 0 && h.scb.State() == StateListen

tcp/handler_test.go

@@ -20,6 +20,103 @@ func TestHandler(t *testing.T) {
 	sendDataFull(t, client, server, []byte("hello"), rawbuf[:])
 }
 
+func TestHandler_RequeueControlRetransmitsSYN(t *testing.T) {
+	const mtu = ethernet.MaxMTU
+	const maxpackets = 3
+	rng := rand.New(rand.NewSource(1))
+	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
+	setupClientServer(t, rng, client, server)
+
+	var rawbuf [mtu]byte
+	n, err := client.Send(rawbuf[:])
+	if err != nil {
+		t.Fatal("client sending initial SYN:", err)
+	} else if n < sizeHeaderTCP {
+		t.Fatalf("initial SYN size=%d, want at least %d", n, sizeHeaderTCP)
+	}
+	initial := mustSegment(t, rawbuf[:n], 0)
+	if initial.Flags != FlagSYN {
+		t.Fatalf("initial flags=%s, want SYN", initial.Flags)
+	}
+	if client.State() != StateSynSent {
+		t.Fatalf("client state=%s, want SYN-SENT", client.State())
+	}
+
+	clear(rawbuf[:])
+	n, err = client.Send(rawbuf[:])
+	if err != nil {
+		t.Fatal("client sending before RequeueControl:", err)
+	} else if n != 0 {
+		t.Fatalf("Send before RequeueControl wrote %d bytes, want 0", n)
+	}
+
+	client.RequeueControl()
+	clear(rawbuf[:])
+	n, err = client.Send(rawbuf[:])
+	if err != nil {
+		t.Fatal("client retransmitting SYN:", err)
+	} else if n < sizeHeaderTCP {
+		t.Fatalf("retransmitted SYN size=%d, want at least %d", n, sizeHeaderTCP)
+	}
+	retransmit := mustSegment(t, rawbuf[:n], 0)
+	if retransmit.Flags != FlagSYN {
+		t.Fatalf("retransmit flags=%s, want SYN", retransmit.Flags)
+	}
+	if retransmit.SEQ != initial.SEQ {
+		t.Fatalf("retransmit SEQ=%d, want initial SEQ=%d", retransmit.SEQ, initial.SEQ)
+	}
+
+	if err := server.Recv(rawbuf[:n]); err != nil {
+		t.Fatal("server receiving retransmitted SYN:", err)
+	}
+	clear(rawbuf[:])
+	n, err = server.Send(rawbuf[:])
+	if err != nil {
+		t.Fatal("server sending SYN-ACK:", err)
+	}
+	segSynAck := mustSegment(t, rawbuf[:n], 0)
+	if segSynAck.Flags != synack {
+		t.Fatalf("server flags=%s, want SYN-ACK", segSynAck.Flags)
+	}
+	if err := client.Recv(rawbuf[:n]); err != nil {
+		t.Fatal("client receiving SYN-ACK:", err)
+	}
+	if client.State() != StateEstablished {
+		t.Fatalf("client state=%s, want ESTABLISHED", client.State())
+	}
+
+	clear(rawbuf[:])
+	n, err = client.Send(rawbuf[:]) // final ACK.
+	if err != nil {
+		t.Fatal("client sending final ACK:", err)
+	}
+	if ack := mustSegment(t, rawbuf[:n], 0); ack.Flags != FlagACK {
+		t.Fatalf("client final flags=%s, want ACK", ack.Flags)
+	}
+	if err := server.Recv(rawbuf[:n]); err != nil {
+		t.Fatal("server receiving final ACK:", err)
+	}
+
+	client.RequeueControl()
+	clear(rawbuf[:])
+	n, err = client.Send(rawbuf[:])
+	if err != nil {
+		t.Fatal("client sending after establishment:", err)
+	} else if n != 0 {
+		seg := mustSegment(t, rawbuf[:n], 0)
+		t.Fatalf("Send after establishment wrote %d bytes (%s), want 0", n, seg.Flags)
+	}
+}
+
+func mustSegment(t *testing.T, b []byte, payloadLen int) Segment {
+	t.Helper()
+	frame, err := NewFrame(b)
+	if err != nil {
+		t.Fatal("parse TCP frame:", err)
+	}
+	return frame.Segment(payloadLen)
+}
+
 func sendDataFull(t *testing.T, client, server *Handler, data, packetBuf []byte) {
 	n, err := client.Write(data)
 	if err != nil {

x/xnet/stack-async.go

@@ -103,6 +103,8 @@ type StackConfig struct {
 	MTU uint16
 	// Accept multicast ethernet and IP packets. Needed for MDNS.
 	AcceptMulticast bool
+	// Accept broadcast IPv4 packets. Needed for managing access points and DHCPv4 servers.
+	AcceptIPv4Broadcast bool
 }
 
 func (cfg *StackConfig) id() uint16 {
@@ -234,7 +236,7 @@ func (s *StackAsync) Reset(cfg StackConfig) (err error) {
 	}
 	s.ip4.SetAddr4(cfg.StaticAddress4)
 	s.setAcceptMulticast4(cfg.AcceptMulticast)
-
+	s.ip4.SetAcceptBroadcast4(cfg.AcceptIPv4Broadcast)
 	s.arpt.passivePeers = uint8(cfg.PassivePeers)
 	err = s.resetARP()
 	if err != nil {

x/xnet/stack-blocking.go

@@ -31,8 +31,20 @@ func (s *StackAsync) StackBlocking(stackProtoBackoff lneto.BackoffStrategy) Stac
 }
 
 type StackBlocking struct {
-	async    *StackAsync
-	_backoff lneto.BackoffStrategy
+	async     *StackAsync
+	_backoff  lneto.BackoffStrategy
+	_nanotime func() int64
+}
+
+func (s StackBlocking) nanotime() int64 {
+	if s._nanotime != nil {
+		return s._nanotime()
+	}
+	return time.Now().UnixNano()
+}
+
+func (s StackBlocking) deadlineTO(timeout time.Duration) int64 {
+	return int64(timeout) + s.nanotime()
 }
 
 func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPResults, error) {
@@ -41,7 +53,7 @@ func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPRe
 		return nil, err
 	}
 	var backoffs uint
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	requested := false
 	var lastState dhcpv4.ClientState
 	for range maxIter {
@@ -108,7 +120,7 @@ func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset
 		return -1, err
 	}
 
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	var done bool
 	var backoffs uint
 	for range maxIter {
@@ -130,7 +142,7 @@ func (s StackBlocking) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D
 		return hw, err
 	}
 	var backoffs uint
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	for range maxIter {
 		hw, err = s.async.ResultResolveHardwareAddress6(addr)
 		if err == nil {
@@ -159,7 +171,7 @@ func (s StackBlocking) DoLookupIPType(host string, timeout time.Duration, qtype
 		return nil, err
 	}
 
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	var backoffs uint
 	for range maxIter {
 		addrs, completed, err := s.async.ResultLookupIP(host)
@@ -181,7 +193,15 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 	if err != nil {
 		return err
 	}
-	deadline := time.Now().Add(timeout)
+	err = s.waitDialTCP(conn, timeout)
+	if err != nil {
+		conn.Abort()
+	}
+	return err
+}
+
+func (s StackBlocking) waitDialTCP(conn *tcp.Conn, timeout time.Duration) (err error) {
+	deadline := s.deadlineTO(timeout)
 	var backoffs uint
 	for range maxIter {
 		state := conn.State()
@@ -189,12 +209,10 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 			return nil
 		} else if state == tcp.StateSynSent || state == tcp.StateSynRcvd || conn.AwaitingSynSend() {
 			if err = s.checkDeadline(deadline); err != nil {
-				conn.Abort()
 				return err
 			}
 		} else {
 			// Unexpected state, abort and terminate connection.
-			conn.Abort()
 			return errTCPFailedToConnect
 		}
 		s.backoff(backoffs)
@@ -203,8 +221,8 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 	return errDeadlineExceed
 }
 
-func (s StackBlocking) checkDeadline(deadline time.Time) error {
-	if time.Since(deadline) > 0 {
+func (s StackBlocking) checkDeadline(deadline int64) error {
+	if s.nanotime() > deadline {
 		return errDeadlineExceed
 	}
 	return nil

x/xnet/stack-go.go

@@ -18,10 +18,15 @@ import (
 const (
 	sockSTREAM = 0x1
 	sockDGRAM  = 0x2
+
+	defaultTCPDialTimeout = 2 * time.Second
+	defaultTCPDialRetries = 1
 )
 
 type StackGoConfig struct {
 	ListenerPoolConfig TCPPoolConfig
+	TCPDialTimeout     time.Duration
+	TCPDialRetries     int
 }
 
 func (s *StackAsync) StackGo(stackProtoBackoff lneto.BackoffStrategy, cfg StackGoConfig) StackGo {
@@ -32,16 +37,27 @@ func (s *StackAsync) StackGo(stackProtoBackoff lneto.BackoffStrategy, cfg StackG
 }
 
 func (s StackBlocking) StackGo(cfg StackGoConfig) StackGo {
+	if cfg.TCPDialRetries <= 0 {
+		cfg.TCPDialRetries = defaultTCPDialRetries
+	}
+	if cfg.TCPDialTimeout <= 0 {
+		cfg.TCPDialTimeout = defaultTCPDialTimeout
+	}
+
 	sg := StackGo{
-		blk:   s,
-		plcfg: cfg.ListenerPoolConfig,
+		blk:            s,
+		plcfg:          cfg.ListenerPoolConfig,
+		tcpDialTimeout: cfg.TCPDialTimeout,
+		tcpDialRetries: cfg.TCPDialRetries,
 	}
 	return sg
 }
 
 type StackGo struct {
-	blk   StackBlocking
-	plcfg TCPPoolConfig
+	blk            StackBlocking
+	plcfg          TCPPoolConfig
+	tcpDialTimeout time.Duration
+	tcpDialRetries int
 }
 
 func (s StackGo) Socket(ctx context.Context, network string, family, sotype int, laddr, raddr net.Addr) (c any, err error) {
@@ -171,7 +187,7 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 			if err != nil {
 				return nil, err
 			}
-			err = s.blk.async.DialTCP(&conn, laddr.Port(), raddr)
+			err = s.blk.StackRetrying().DoDialTCP(&conn, laddr.Port(), raddr, s.tcpDialTimeout, s.tcpDialRetries)
 			if err != nil {
 				return nil, err
 			}

x/xnet/stack-retrying.go

@@ -13,9 +13,11 @@ func (s *StackAsync) StackRetrying(stackProtoBackoff lneto.BackoffStrategy) Stac
 	if stackProtoBackoff == nil {
 		panic("nil backoff to StackRetrying")
 	}
-	return StackRetrying{
-		block: s.StackBlocking(stackProtoBackoff),
-	}
+	return s.StackBlocking(stackProtoBackoff).StackRetrying()
+}
+
+func (s StackBlocking) StackRetrying() StackRetrying {
+	return StackRetrying{block: s}
 }
 
 var (
@@ -93,13 +95,28 @@ func (s StackRetrying) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D
 func (s StackRetrying) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.AddrPort, timeout time.Duration, retries int) (err error) {
 	expectEnd := time.Now().Add(timeout * time.Duration(retries))
 	var firstErr error
-	for range retries {
-		err = s.block.DoDialTCP(conn, localPort, addrp, timeout)
+	for i := range retries {
+		if i == 0 || conn.State().IsClosed() {
+			err = s.block.async.DialTCP(conn, localPort, addrp)
+			if err != nil {
+				if firstErr == nil {
+					firstErr = err
+				}
+				continue
+			}
+		} else if conn.IsAwaitingControl() {
+			conn.RequeueControl()
+		}
+
+		err = s.block.waitDialTCP(conn, timeout)
 		if err == nil {
 			return nil
 		} else if firstErr == nil {
 			firstErr = err
 		}
+		if !conn.IsAwaitingControl() {
+			conn.Abort()
+		}
 	}
 	if time.Now().Before(expectEnd) {
 		if err != firstErr {

x/xnet/xnet_test.go

@@ -2,10 +2,12 @@ package xnet
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"math/rand"
 	"net/netip"
 	"sync"
+	"syscall"
 	"testing"
 	"time"
 
@@ -28,6 +30,57 @@ const (
 	finack = tcp.FlagFIN | tcp.FlagACK
 )
 
+func newstackTestScheduler(t testing.TB) stackTestScheduler {
+	return stackTestScheduler{
+		t:                   t,
+		stackBackoffSignal:  make(chan struct{}),
+		stackContinueSignal: make(chan struct{}),
+		timeout:             time.Second,
+	}
+}
+
+type stackTestScheduler struct {
+	t testing.TB
+	// when stack backs off it signals here and waits until channel read or timeout.
+	stackBackoffSignal chan struct{}
+	// when main goroutine is ready for more information this channel is written to to signal waiting on stack activity.
+	stackContinueSignal chan struct{}
+	timeout             time.Duration
+}
+
+func (ss *stackTestScheduler) backoffStack(consecutiveBackoffs uint) time.Duration {
+	timeout := time.After(ss.timeout)
+	select {
+	case ss.stackBackoffSignal <- struct{}{}:
+	case <-timeout:
+		ss.t.Fatal("timeout backing off, possible race condition? Multiple stacks using same backoff is unexpected pattern")
+	}
+	select {
+	case <-ss.stackContinueSignal:
+	case <-timeout:
+		ss.t.Fatal("timeout waiting for continue")
+	}
+	return lneto.BackoffFlagNop // backoff yield implemented on our side.
+}
+
+func (ss *stackTestScheduler) mainGoroutineWaitForStackYield() {
+	timeout := time.After(ss.timeout)
+	select {
+	case <-ss.stackBackoffSignal:
+	case <-timeout:
+		ss.t.Fatal("timeout waiting for stack to backoff")
+	}
+}
+
+func (ss *stackTestScheduler) mainGoroutineYieldToStack() {
+	timeout := time.After(ss.timeout)
+	select {
+	case ss.stackContinueSignal <- struct{}{}:
+	case <-timeout:
+		ss.t.Fatal("timeout while trying to yield to stack")
+	}
+}
+
 func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) {
 	const seed = 5678
 	const MTU = ethernet.MaxMTU
@@ -105,6 +158,78 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) {
 	}
 }
 
+func TestStackGoTCPDialRetriesPendingControl(t *testing.T) {
+	const seed = 5678
+	const MTU = ethernet.MaxMTU
+	const tcptimeout = time.Second
+	const yield = 1 * time.Millisecond
+	client, sv, _, _ := newTCPStacks(t, seed, MTU)
+	tbackoffer := newstackTestScheduler(t)
+
+	sg := client.StackBlocking(tbackoffer.backoffStack).StackGo(StackGoConfig{
+		ListenerPoolConfig: TCPPoolConfig{
+			QueueSize: 4,
+			TxBufSize: MTU,
+			RxBufSize: MTU,
+			NewBackoff: func() lneto.BackoffStrategy {
+				return backoffYield
+			},
+		},
+		TCPDialTimeout: tcptimeout,
+		TCPDialRetries: 2,
+	})
+	t.Log("start")
+	// closure to simulate time.
+	var now time.Duration
+	sg.blk._nanotime = func() int64 { return int64(now) }
+
+	laddr := netip.AddrPortFrom(netip.AddrFrom4(client.Addr4()), 1234)
+	raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), 22)
+	done := make(chan error, 1)
+	go func() {
+		_, err := sg.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, laddr, raddr)
+		done <- err
+	}()
+	npacket := 0
+	ntcppacket := 0
+	var buf [ethernet.MaxMTU + ethernet.MaxOverheadSize]byte
+	for !t.Failed() { // Tinygo does not implement failnow.
+		tbackoffer.mainGoroutineWaitForStackYield()
+		n, err := client.EgressEthernet(buf[:])
+		now += tcptimeout / 100
+		if err != nil {
+			t.Fatal(err)
+		} else if n == 0 {
+			t.Fatal("expected packet from socketnetip", npacket, ntcppacket)
+		}
+		npacket++
+		frm, ok := getTCPFrame(buf[:])
+		if !ok {
+			tbackoffer.mainGoroutineYieldToStack()
+			continue
+		}
+		ntcppacket++
+		_, flags := frm.OffsetAndFlags()
+		if flags != tcp.FlagSYN {
+			t.Fatal("expected SYN packet")
+		}
+		switch ntcppacket {
+		case 1:
+			now += 2 * tcptimeout
+			tbackoffer.mainGoroutineYieldToStack()
+		case 2:
+			now += 2 * tcptimeout
+			tbackoffer.mainGoroutineYieldToStack()
+			select {
+			case <-time.After(time.Second):
+				t.Fatal("SocketNetip hanging")
+			case <-done:
+				return // Test success.
+			}
+		}
+	}
+}
+
 func TestStackAsyncTCP_multipacket(t *testing.T) {
 	const seed = 1234
 	const MTU = 512