Ack dropping - fix for #50 (#127)

* fix for #50

* fix remaining test

* revert changes and instead approach problem on full-duplex case

* fix merge messup

* fix @MDr164 note and upgrade documentation while at it

* gate full buffer exit on rx shutdown since no risk of full buffer error

* add log line to note issue is hit

* document CloseRead relationship with Close

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()
 }

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),

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
 }

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