diff --git a/tcp/conn.go b/tcp/conn.go index 190fedd..5aac264 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -51,16 +51,25 @@ func (conn *Conn) reset(h Handler) { conn.ipID = 0 } +// ConnConfig provides configuration parameters for [Conn]. type ConnConfig struct { - RxBuf []byte - TxBuf []byte + RxBuf []byte // Fixed size buffer for incoming data via [Conn.Read]. + TxBuf []byte // Fixed size buffer for egress data via [Conn.Write]. + // TxPacketQueueSize is the maximum number of sent-but-unacknowledged segments + // the connection tracks at once for retransmission. Each queued entry records the + // sequence range of one outgoing segment and is released when the peer ACKs it; + // once the queue is full no further segments are emitted until an ACK frees a slot. + // Must be greater than zero and no larger than len(TxBuf). TxPacketQueueSize int // RWBackoff sets the backoff policy for backoff when data unavailable on Read or buffer full on Write. // If not set a default backoff strategy will be used. See [internal.BackoffConnRW]. RWBackoff lneto.BackoffStrategy - Logger *slog.Logger + // Logger sets the [Conn] logger. + // Lower level logging available at [Handler.SetLoggers] via [Conn.InternalHandler]. + Logger *slog.Logger } +// Configure should be called on any newly created connection before usage. See [ConnConfig]. func (conn *Conn) Configure(config ConnConfig) (err error) { if config.RWBackoff == nil { return lneto.ErrMissingHALConfig @@ -90,6 +99,7 @@ func (conn *Conn) RemotePort() uint16 { return conn.h.RemotePort() } +// RemoteAddr returns the address of the peer Conn is exchanging data with. func (conn *Conn) RemoteAddr() []byte { conn.mu.Lock() defer conn.mu.Unlock() @@ -189,6 +199,7 @@ func (conn *Conn) OpenListen(localPort uint16, iss Value) error { // CloseRead activates local discard mode on the connection. Incoming data is // still ACKed normally but payload is dropped; future Read calls return io.EOF. // The write side is unaffected. +// If [Conn.Close] is also called the connection is terminated. func (conn *Conn) CloseRead() error { conn.mu.Lock() defer conn.mu.Unlock() @@ -200,6 +211,7 @@ func (conn *Conn) CloseRead() error { return nil } +// Close will initiate TCP close sequence. After Close is called future [Conn.Write] calls will fail with [net.ErrClosed]. func (conn *Conn) Close() error { conn.mu.Lock() defer conn.mu.Unlock() @@ -272,6 +284,7 @@ func (conn *Conn) Write(b []byte) (int, error) { return n, err } +// Flush blocks until all buffered TCP data has been sent. func (conn *Conn) Flush() error { connid, err := conn.lockPipeConnID() if err != nil { @@ -371,6 +384,7 @@ func (conn *Conn) checkPipeOpen() error { return nil } +// Demux implements [lneto.StackNode]. func (conn *Conn) Demux(buf []byte, off int) (err error) { conn.mu.Lock() defer conn.mu.Unlock() @@ -398,6 +412,7 @@ func (conn *Conn) Demux(buf []byte, off int) (err error) { return nil } +// Encapsulate implements [lneto.StackNode]. func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (n int, err error) { conn.mu.Lock() defer conn.mu.Unlock() @@ -427,6 +442,7 @@ func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) return n, nil } +// Protocol implements [lneto.StackNode]. func (conn *Conn) Protocol() uint64 { return uint64(lneto.IPProtoTCP) } @@ -491,6 +507,7 @@ func (conn *Conn) deadlineExceeded(deadline *time.Time) bool { return !deadline.IsZero() && time.Since(*deadline) > 0 } +// ConnectionID implements [lneto.StackNode]. func (conn *Conn) ConnectionID() *uint64 { return conn.h.ConnectionID() } diff --git a/tcp/control.go b/tcp/control.go index e79023f..0effb55 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -146,6 +146,15 @@ func (tcb *ControlBlock) MakeKeepalive() Segment { } } +// QueueRST queues a RST segment to be emitted on the next send, overriding any +// other pending flags. seq is the sequence number the RST will carry (per RFC +// 9293 reset generation, the acknowledged value SEG.ACK of the offending +// segment). The connection should be torn down once the RST is sent. +func (tcb *ControlBlock) QueueRST(seq Value) { + tcb.pending = [2]Flags{0: FlagRST} + tcb.rstPtr = seq +} + // MakeDupACK returns a duplicate ACK segment suitable for fast-retransmit // recovery signaling, without advancing the sender ACK boundary. Useful for: // - constructing an explicit duplicate ACK from local state (e.g. test harness), diff --git a/tcp/handler.go b/tcp/handler.go index 34dea8f..ae0510f 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -14,7 +14,7 @@ import ( // related to data buffering, frame sequencing and connection state handling. // Does NOT implement IP related logic, so no CRC calculation/validation or pseudo header logic. // -// See [Conn] for a higher level abstraction of a TCP connection, and see [ControlBlock] for the lower level bits of a TCP connection. +// See [Conn] for a higher level abstraction of a TCP connection, and see [ControlBlock] for the low level state machine of a TCP connection. type Handler struct { connid uint64 scb ControlBlock @@ -35,6 +35,7 @@ type Handler struct { nRetransmit uint8 } +// SetLoggers sets the [slog.Logger] for the Handler and internal [ControlBlock]. func (h *Handler) SetLoggers(handler, scb *slog.Logger) { h.logger.log = handler h.scb.logger.log = scb @@ -118,6 +119,7 @@ func (h *Handler) Abort() { h.reset(0, 0, 0) } +// reset clears all state except [ControlBlock] state. So [Handler.State] will remain unchanged. func (h *Handler) reset(localPort, remotePort uint16, iss Value) { *h = Handler{ connid: h.connid + 1, @@ -160,7 +162,7 @@ func (h *Handler) Recv(incomingPacket []byte) error { return lneto.ErrMismatch } payload := tfrm.Payload() - if len(payload) > h.bufRx.Free() { + if !h.shutdownRx && len(payload) > h.bufRx.Free() { return lneto.ErrBufferFull } segIncoming := tfrm.Segment(len(payload)) @@ -186,6 +188,15 @@ func (h *Handler) Recv(incomingPacket []byte) error { if prevState != h.scb.State() { h.info("tcp.Handler:rx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("old", prevState.String()), slog.String("new", h.scb.State().String()), slog.String("rxflags", segIncoming.Flags.String())) } + if segIncoming.DATALEN != 0 && h.shutdownRx && (h.scb.State() == StateFinWait1 || h.scb.State() == StateFinWait2) { + // soypat/lneto#50: the application is done in both directions — read side + // shut down (CloseRead) and our FIN sent (Close) — so inbound data has no + // consumer. Reply RST instead of the silent ACK-and-drop that leaves the + // peer waiting; the connection is torn down once the RST is sent. + h.info("tcp.Handler:rst-data-after-fullclose", slog.Uint64("lport", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort)), slog.Uint64("datalen", uint64(segIncoming.DATALEN))) + h.scb.QueueRST(segIncoming.ACK) + return nil + } if segIncoming.DATALEN != 0 && !h.shutdownRx { _, err = h.bufRx.Write(payload) if err != nil { @@ -231,12 +242,16 @@ func (h *Handler) Recv(incomingPacket []byte) error { } // ShutdownRead activates local discard mode: incoming payload bytes are dropped -// (ACK/SEQ still advance normally) and Read returns io.EOF immediately. -// Not reversible within the lifetime of a connection; reset clears it. +// (ACK/SEQ still advance normally) and Read returns [io.EOF] immediately. +// Not reversible within the lifetime of a connection. +// If [Handler.Close] and this method are both called then connection will be terminated. func (h *Handler) ShutdownRead() { h.shutdownRx = true } +// Close will initiate the TCP close sequence. +// After Close is called [Handler.Write] will fail with [net.ErrClosed]. +// The connection may still receive data to read after Close called. func (h *Handler) Close() error { h.trace("tcp.Handler.Close") if h.closing { @@ -328,6 +343,10 @@ func (h *Handler) Send(b []byte) (int, error) { closedSuccess := prevState == StateTimeWait && segment.Flags.HasAny(FlagACK) if closedSuccess { h.reset(0, 0, 0) + } else if segment.Flags.HasAny(FlagRST) { + // A sent RST aborts the connection: tear down local state now that the + // reset has been written to the wire (frame already in b). + h.Abort() } return datalen, nil } diff --git a/tcp/handler_test.go b/tcp/handler_test.go index 1fa97af..c411cbf 100644 --- a/tcp/handler_test.go +++ b/tcp/handler_test.go @@ -1155,6 +1155,146 @@ func TestHandler_RetransmitAfterMultipleLossesBothDirections(t *testing.T) { } } +// driveToFinWait2 performs an active close from client: Close() → FIN, the +// server ACKs it (no FIN of its own), leaving client in FIN-WAIT-2. +func driveToFinWait2(t *testing.T, client, server *Handler, buf []byte) { + t.Helper() + if err := client.Close(); err != nil { + t.Fatal("client close:", err) + } + clear(buf) + n, err := client.Send(buf) + if err != nil { + t.Fatal("client send FIN:", err) + } + if client.State() != StateFinWait1 { + t.Fatal("client not FIN-WAIT-1:", client.State()) + } + if err := server.Recv(buf[:n]); err != nil { + t.Fatal("server recv FIN:", err) + } + clear(buf) + n, err = server.Send(buf) // pure ACK of the FIN. + if err != nil { + t.Fatal("server send ACK:", err) + } + if err := client.Recv(buf[:n]); err != nil { + t.Fatal("client recv ACK:", err) + } + if client.State() != StateFinWait2 { + t.Fatal("client not FIN-WAIT-2:", client.State()) + } +} + +// serverSendData writes data on the server and emits it as one packet, returning +// the packet length in buf. +func serverSendData(t *testing.T, server *Handler, data, buf []byte) int { + t.Helper() + if _, err := server.Write(data); err != nil { + t.Fatal("server write:", err) + } + clear(buf) + n, err := server.Send(buf) + if err != nil { + t.Fatal("server send data:", err) + } + if n <= sizeHeaderTCP { + t.Fatal("server emitted no data segment") + } + return n +} + +// TestFinWait2_FullClose_RST: when the application is done in BOTH directions — +// read side shut down ([Handler.ShutdownRead]) AND our FIN sent (Close) — inbound +// data in FIN-WAIT-2 has no consumer. The Handler must reply RST (not silently +// ACK-and-drop, which leaves the peer waiting) and tear down the local +// connection. Regression for soypat/lneto#50, reworked to gate on the read +// shutdown so RFC half-close is preserved (see TestFinWait2_HalfClose_DataReadable). +func TestFinWait2_FullClose_RST(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(50)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + client.ShutdownRead() // application is done reading. + driveToFinWait2(t, client, server, buf[:]) + + // Peer pipelines a request on the fully-closed connection. + n := serverSendData(t, server, []byte("GET / HTTP/1.1\r\nHost: x\r\n\r\n"), buf[:]) + err := client.Recv(buf[:n]) + if err != nil && !IsDroppedErr(err) { + t.Fatal("client recv data:", err) + } + + // Response must be RST. + clear(buf[:]) + n, err = client.Send(buf[:]) + if err != nil { + t.Fatal("client send response:", err) + } + if n < sizeHeaderTCP { + t.Fatal("no response emitted to data in fully-closed FIN-WAIT-2") + } + resp, _ := NewFrame(buf[:n]) + if !resp.Segment(0).Flags.HasAny(FlagRST) { + t.Fatalf("data after full close must elicit RST; got flags=%s", resp.Segment(0).Flags) + } + + // Post-RST teardown: the local connection must be gone, not stuck in FIN-WAIT-2. + if client.State() != StateClosed { + t.Fatalf("connection not torn down after RST: state=%s (want CLOSED)", client.State()) + } +} + +// TestFinWait2_HalfClose_DataReadable guards the RFC half-close model: when only +// Close() was called (write side done) but the read side is still open, data +// arriving in FIN-WAIT-2 must be ACKed and delivered to the application — never +// reset. This is the case the original #50 fix wrongly broke. +func TestFinWait2_HalfClose_DataReadable(t *testing.T) { + const mtu = ethernet.MaxMTU + rng := rand.New(rand.NewSource(51)) + client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3) + setupClientServer(t, rng, client, server) + var buf [mtu]byte + establish(t, client, server, buf[:]) + + // NOTE: no ShutdownRead — app closed the write half only, still reading. + driveToFinWait2(t, client, server, buf[:]) + + data := []byte("late peer data") + n := serverSendData(t, server, data, buf[:]) + if err := client.Recv(buf[:n]); err != nil { + t.Fatalf("half-close: data in FIN-WAIT-2 must be accepted, not dropped: %v", err) + } + if client.State() != StateFinWait2 { + t.Fatalf("state changed on half-close data: got %s want FIN-WAIT-2", client.State()) + } + + // Data must be readable by the application. + var rd [64]byte + rn, err := client.Read(rd[:]) + if err != nil { + t.Fatal("client read:", err) + } + if string(rd[:rn]) != string(data) { + t.Fatalf("read %q; want %q", rd[:rn], data) + } + + // The response must never be a RST. + clear(buf[:]) + n, err = client.Send(buf[:]) + if err != nil { + t.Fatal("client send:", err) + } + if n >= sizeHeaderTCP { + if s, _ := NewFrame(buf[:n]); s.Segment(0).Flags.HasAny(FlagRST) { + t.Fatal("half-close reader must not RST inbound data") + } + } +} + // TestRetransmit_CumulativeACK_NoSpurious reproduces soypat/lneto#57. // lneto streams 4 segments (TX queue=4); the first is "lost". A Linux-style // remote buffers the rest out of order and dup-ACKs the hole. After 3 dup ACKs