continue adding Handler logic; remove unused TxQueue tests

2026-08-16 21:03:25 +00:00 · 2025-02-09 23:43:30 -03:00
parent aa8da93f6b
commit 64a7102064
9 changed files with 520 additions and 132 deletions
@@ -53,7 +53,7 @@ type ControlBlock struct {
 	pending      [2]Flags
 	state        State
 	challengeAck bool
-	log          *slog.Logger
+	logger
 }

 // State returns the current state of the TCP connection.
@@ -88,7 +88,7 @@ func (tcb *ControlBlock) SetRecvWindow(wnd Size) {

 // SetLogger sets the logger to be used by the ControlBlock.
 func (tcb *ControlBlock) SetLogger(log *slog.Logger) {
-	tcb.log = log
+	tcb.logger = logger{log: log}
 }

 // IncomingIsKeepalive checks if an incoming segment is a keepalive segment.
@@ -1,30 +1,34 @@
 package tcp

 import (
-	"context"
 	"log/slog"

 	"github.com/soypat/lneto/internal"
 )

-func (tcb *ControlBlock) logenabled(lvl slog.Level) bool {
-	return internal.HeapAllocDebugging || (tcb.log != nil && tcb.log.Handler().Enabled(context.Background(), lvl))
+// logger provides methods that can be easily attached by struct embedding
+// to types in this package.
+type logger struct {
+	log *slog.Logger
 }

-func (tcb *ControlBlock) logattrs(lvl slog.Level, msg string, attrs ...slog.Attr) {
-	internal.LogAttrs(tcb.log, lvl, msg, attrs...)
+func (l logger) logerr(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, slog.LevelError, msg, attrs...)
 }
-
-func (tcb *ControlBlock) debug(msg string, attrs ...slog.Attr) {
-	tcb.logattrs(slog.LevelDebug, msg, attrs...)
+func (l logger) info(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, slog.LevelInfo, msg, attrs...)
 }
-
-func (tcb *ControlBlock) trace(msg string, attrs ...slog.Attr) {
-	tcb.logattrs(internal.LevelTrace, msg, attrs...)
+func (l logger) warn(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, slog.LevelWarn, msg, attrs...)
 }
-
-func (tcb *ControlBlock) logerr(msg string, attrs ...slog.Attr) {
-	tcb.logattrs(slog.LevelError, msg, attrs...)
+func (l logger) debug(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, slog.LevelDebug, msg, attrs...)
+}
+func (l logger) trace(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, internal.LevelTrace, msg, attrs...)
+}
+func (l logger) logenabled(lvl slog.Level) bool {
+	return internal.LogEnabled(l.log, lvl)
 }

 func (tcb *ControlBlock) traceSnd(msg string) {
@@ -0,0 +1,198 @@
+package tcp
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"math"
+
+	"github.com/soypat/lneto/lneto2"
+)
+
+var (
+	errShortTCP    = errors.New("TCP offset exceeds frame")
+	errBadTCPOff   = errors.New("TCP offset invalid")
+	errEvilPacket  = errors.New("evil packet")
+	errZeroDstPort = errors.New("TCP zero destination port")
+	errZeroSrcPort = errors.New("TCP zero source port")
+)
+
+const (
+	sizeHeaderTCP = 20
+)
+
+// NewFrame returns a new TCPFrame with data set to buf.
+// An error is returned if the buffer size is smaller than 20.
+// Users should still call [Frame.ValidateSize] before working
+// with payload/options of frames to avoid panics.
+func NewFrame(buf []byte) (Frame, error) {
+	if len(buf) < sizeHeaderTCP {
+		return Frame{buf: nil}, errors.New("TCP packet too short")
+	}
+	return Frame{buf: buf}, nil
+}
+
+// Frame encapsulates the raw data of a TCP segment
+// and provides methods for manipulating, validating and
+// retrieving fields and payload data. See [RFC9293].
+//
+// [RFC9293]: https://datatracker.ietf.org/doc/html/rfc9293
+type Frame struct {
+	buf []byte
+}
+
+// RawData returns the underlying slice with which the frame was created.
+func (tfrm Frame) RawData() []byte { return tfrm.buf }
+
+// SourcePort identifies the sending port of the TCP packet. Must be non-zero.
+func (tfrm Frame) SourcePort() uint16 {
+	return binary.BigEndian.Uint16(tfrm.buf[0:2])
+}
+
+// SetSourcePort sets TCP source port. See [Frame.SetSourcePort]
+func (tfrm Frame) SetSourcePort(src uint16) {
+	binary.BigEndian.PutUint16(tfrm.buf[0:2], src)
+}
+
+// DestinationPort identifies the receiving port for the TCP packet. Must be non-zero.
+func (tfrm Frame) DestinationPort() uint16 {
+	return binary.BigEndian.Uint16(tfrm.buf[2:4])
+}
+
+// SetDestinationPort sets TCP destination port. See [Frame.DestinationPort]
+func (tfrm Frame) SetDestinationPort(dst uint16) {
+	binary.BigEndian.PutUint16(tfrm.buf[2:4], dst)
+}
+
+// Seq returns sequence number of the first data octet in this segment (except when SYN present)
+// If SYN present this is the Initial Sequence Number (ISN) and the first data octet would be ISN+1.
+func (tfrm Frame) Seq() Value {
+	return Value(binary.BigEndian.Uint32(tfrm.buf[4:8]))
+}
+
+// SetSeq sets Seq field. See [Frame.Seq].
+func (tfrm Frame) SetSeq(v Value) {
+	binary.BigEndian.PutUint32(tfrm.buf[4:8], uint32(v))
+}
+
+// Ack is the next sequence number (Seq field) the sender is expecting to receive (when ACK is present).
+// In other words an Ack of X indicates all octets up to but not including X have been received.
+// Once a connection is established the ACK flag should always be set.
+func (tfrm Frame) Ack() Value {
+	return Value(binary.BigEndian.Uint32(tfrm.buf[8:12]))
+}
+
+// SetAck sets Ack field. See [Frame.Ack].
+func (tfrm Frame) SetAck(v Value) {
+	binary.BigEndian.PutUint32(tfrm.buf[8:12], uint32(v))
+}
+
+// OffsetAndFlags returns the offset and flag fields of TCP header.
+// Offset is amount of 32-bit words used for TCP header including TCP options (see [Frame.HeaderLength]).
+// See [Flags] for more information on TCP flags.
+func (tfrm Frame) OffsetAndFlags() (offset uint8, flags Flags) {
+	v := binary.BigEndian.Uint16(tfrm.buf[12:14])
+	offset = uint8(v >> 12)
+	flags = Flags(v).Mask()
+	return offset, flags
+}
+
+// SetOffsetAndFlags returns offset and flag fields of TCP header. See [Frame.OffsetAndFlags].
+func (tfrm Frame) SetOffsetAndFlags(offset uint8, flags Flags) {
+	v := uint16(offset)<<12 | uint16(flags.Mask())
+	binary.BigEndian.PutUint16(tfrm.buf[12:14], v)
+}
+
+// HeaderLength uses Offset field to calculate the total length of
+// the TCP header including options. Performs no validation.
+func (tfrm Frame) HeaderLength() (tcpWords int) {
+	offset, _ := tfrm.OffsetAndFlags()
+	return 4 * int(offset)
+}
+
+func (tfrm Frame) WindowSize() uint16 { return binary.BigEndian.Uint16(tfrm.buf[14:16]) }
+func (tfrm Frame) SetWindowSize(v uint16) {
+	binary.BigEndian.PutUint16(tfrm.buf[14:16], v)
+}
+
+// CRC returns the checksum field in the TCP header.
+func (tfrm Frame) CRC() uint16 {
+	return binary.BigEndian.Uint16(tfrm.buf[16:18])
+}
+
+// SetCRC sets the checksum field of the TCP header. See [Frame.CRC].
+func (tfrm Frame) SetCRC(checksum uint16) {
+	binary.BigEndian.PutUint16(tfrm.buf[16:18], checksum)
+}
+
+func (tfrm Frame) UrgentPtr() uint16      { return binary.BigEndian.Uint16(tfrm.buf[18:20]) }
+func (tfrm Frame) SetUrgentPtr(up uint16) { binary.BigEndian.PutUint16(tfrm.buf[18:20], up) }
+
+// Payload returns the payload content section of the TCP packet (not including TCP options).
+// Be sure to call [Frame.ValidateSize] beforehand to avoid panic.
+func (tfrm Frame) Payload() []byte {
+	return tfrm.buf[tfrm.HeaderLength():]
+}
+
+// Segment returns the [Segment] representation of the TCP header and data length.
+func (tfrm Frame) Segment(payloadSize int) Segment {
+	if payloadSize > math.MaxUint32 {
+		panic("TCP overflow payload size")
+	}
+	return Segment{
+		SEQ:     tfrm.Seq(),
+		ACK:     tfrm.Ack(),
+		WND:     Size(tfrm.WindowSize()),
+		DATALEN: Size(payloadSize),
+		Flags:   Flags(binary.BigEndian.Uint16(tfrm.buf[12:14])).Mask(),
+	}
+}
+
+// Options returns the TCP option buffer portion of the frame. The returned slice may be zero length.
+// Be sure to call [Frame.ValidateSize] beforehand to avoid panic.
+func (tfrm Frame) Options() []byte {
+	return tfrm.buf[sizeHeaderTCP:tfrm.HeaderLength()]
+}
+
+// ClearHeader zeros out the fixed(non-variable) header contents.
+func (frm Frame) ClearHeader() {
+	for i := range frm.buf[:sizeHeaderTCP] {
+		frm.buf[i] = 0
+	}
+}
+
+func (tfrm Frame) String() string {
+	seg := tfrm.Segment(len(tfrm.Payload()))
+	return fmt.Sprintf("%+v", seg)
+}
+
+//
+// Validation API
+//
+
+// func (tfrm Frame) Validate(v *lneto2.Validator) {
+// tfrm.ValidateSize(v)
+// tfrm.ValidateExceptCRC(v)
+// }
+
+// ValidateSize checks the frame's size fields and compares with the actual buffer
+// the frame. It returns a non-nil error on finding an inconsistency.
+func (tfrm Frame) ValidateSize(v *lneto2.Validator) {
+	off := tfrm.HeaderLength()
+	if off < sizeHeaderTCP {
+		v.AddBitPosErr(12*8, 4, errBadTCPOff)
+	}
+	if off > len(tfrm.RawData()) {
+		v.AddBitPosErr(12*8, 4, errShortTCP)
+	}
+}
+
+func (tfrm Frame) ValidateExceptCRC(v *lneto2.Validator) {
+	tfrm.ValidateSize(v)
+	if tfrm.DestinationPort() == 0 {
+		v.AddBitPosErr(2*8, 16, errZeroDstPort)
+	}
+	if tfrm.SourcePort() == 0 {
+		v.AddBitPosErr(0, 16, errZeroSrcPort)
+	}
+}
@@ -0,0 +1,132 @@
+package tcp
+
+import (
+	"errors"
+	"net"
+
+	"log/slog"
+
+	"github.com/soypat/lneto/internal"
+	"github.com/soypat/lneto/lneto2"
+)
+
+var (
+	errMismatchedPort = errors.New("mismatched port")
+)
+
+// Handler is a low level TCP handling data structure. It implements logic
+// 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.
+// Does NOT implement connection lifetime handling, so NO deadlines, keepalives, backoffs or anything that requires use of time package.
+type Handler struct {
+	scb        ControlBlock
+	bufTx      ringTx
+	bufRx      internal.Ring
+	localPort  uint16
+	remotePort uint16
+	// connid is a conenction counter that is incremented each time a new
+	// connection is established via Open calls. This disambiguate's whether
+	// Read and Write calls belong to the current connection.
+	connid    uint8
+	closing   bool
+	validator lneto2.Validator
+	logger
+}
+
+func (h *Handler) Reset() error {
+	*h = Handler{
+		connid: h.connid + 1,
+		bufTx:  h.bufTx,
+		bufRx:  h.bufRx,
+	}
+	h.bufRx.Reset()
+	h.bufTx.ResetOrReuse(nil, 0, 0)
+	return nil
+}
+
+func (h *Handler) Recv(b []byte) error {
+	if h.isClosed() {
+		return net.ErrClosed
+	}
+	tfrm, err := NewFrame(b)
+	if err != nil {
+		return err
+	}
+	tfrm.ValidateExceptCRC(&h.validator)
+	err = h.validator.Err()
+	if err != nil {
+		return err
+	}
+	remotePort := tfrm.SourcePort()
+	if h.remotePort != 0 && remotePort != h.remotePort {
+		return errMismatchedPort
+	}
+	dstPort := tfrm.DestinationPort()
+	if h.localPort != dstPort {
+		return errMismatchedPort
+	}
+	payload := tfrm.Payload()
+	if len(payload) > h.bufRx.Free() {
+		return errors.New("rx buffer full")
+	}
+	segIncoming := tfrm.Segment(len(payload))
+	if h.scb.IncomingIsKeepalive(segIncoming) {
+		h.info("tcp.Handler:rx-keepalive", slog.Uint64("port", uint64(h.localPort)))
+		return nil
+	}
+	prevState := h.scb.State()
+	err = h.scb.Recv(segIncoming)
+	if err != nil {
+		if h.scb.State() == StateClosed {
+			// TODO(soypat): Should return EOF/ErrClosed?
+			err = err // Connection closed by reset.
+		}
+		return err
+	}
+	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 {
+		_, err = h.bufRx.Write(payload)
+		if err != nil {
+			return err
+		}
+	}
+	if segIncoming.Flags.HasAny(FlagSYN) && h.remotePort == 0 {
+		// Remote reached out and has given us their port, set it on our side.
+		h.debug("tcp.Handler:rx-remoteport-set", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("remoteport", uint64(remotePort)))
+		h.remotePort = remotePort
+	}
+	return nil
+}
+
+func (h *Handler) Handle(b []byte) (int, error) {
+	h.trace("tcp.Handler:start", slog.Uint64("port", uint64(h.localPort)))
+	if h.isClosed() {
+		return 0, net.ErrClosed
+	} else if h.AwaitingSyn() {
+		return h.sendInitSyn(b)
+	}
+	tfrm, err := NewFrame(b)
+	if err != nil {
+		return 0, err
+	}
+
+	tfrm.SetSourcePort(h.localPort)
+	tfrm.SetDestinationPort(h.remotePort)
+
+	return 0, nil
+}
+
+func (h *Handler) sendInitSyn(b []byte) (int, error) {
+	return 0, nil
+}
+
+// AwaitingSyn checks if the Handler is waiting for a Syn to arrive.
+func (h *Handler) AwaitingSyn() bool {
+	return h.remotePort != 0 && h.scb.State() == StateSynSent
+}
+
+func (h *Handler) isClosed() bool {
+	return h.closing || h.scb.State().IsClosed()
+}
@@ -70,7 +70,21 @@ func TestTxQueue(t *testing.T) {
 						}
 						currentAck = Add(currentAck, Size(len(msg)))
 					}
+					sent := rtx.BufferedSent()
+					unsent := rtx.Buffered()
+					wantUnsent := int(Add(currentAck, -startAck))
+					if unsent != wantUnsent {
+						t.Fatalf("want %d data buffered, got %d", wantUnsent, unsent)
+					} else if sent != 0 {
+						t.Fatalf("want no data sent, got %d", sent)
+					}
 					operateOnRing(t, &rtx, nil, readBuf[:], aux[:], &currentAck)
+					unsent = rtx.Buffered()
+					if unsent != 0 {
+						t.Fatalf("expected all data to be sent after ack of most recent packet, %d", unsent)
+					} else if rtx.BufferedSent() != 0 {
+						t.Fatal("unexpected buffer not completely acked")
+					}
 				}
 			},
 		},
@@ -83,122 +97,6 @@ func TestTxQueue(t *testing.T) {
 	}
 }

-func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []byte, maxPkt int, startAck Value) {
-	if len(msgs) > maxPkt {
-		panic("need ring buffer to contain messages")
-	}
-	err := rtx.Reset(buf, maxPkt, startAck)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	prevSeq := Value(startAck)
-	packets := make([][]byte, len(msgs))
-	sent := 0
-	for i := range aux {
-		aux[i] = 0
-	}
-	for i, msg := range msgs {
-		if len(aux) < len(msg) {
-			panic("need aux to contain message")
-		}
-		n, err := rtx.Write(msg)
-		if err != nil {
-			t.Fatalf("writing packet %d: %s", i, err)
-		} else if n != len(msg) {
-			t.Fatalf("want %d written, got %d", len(msg), n)
-		}
-		testQueueSanity(t, rtx)
-		unsent := rtx.Buffered()
-		if unsent != n {
-			t.Fatalf("want unsent %d, got %d", n, unsent)
-		}
-		testQueueSanity(t, rtx)
-		n, seq, err := rtx.MakePacket(aux[sent : sent+len(msg)])
-		if err != nil {
-			t.Fatal("MakePacket: ", err)
-		} else if seq != prevSeq {
-			t.Fatalf("want seq %d, got %d", prevSeq, seq)
-		} else if n != len(msg) {
-			t.Fatalf("want full message %d sent, got %d", len(msg), n)
-		}
-		testQueueSanity(t, rtx)
-		gotSent := rtx.BufferedSent()
-		if gotSent != sent+n {
-			t.Fatalf("want sent %d, got %d", sent+n, gotSent)
-		}
-		testQueueSanity(t, rtx)
-		packets = append(packets, aux[sent:sent+n])
-		prevSeq += Value(n)
-		sent += n
-	}
-}
-
-func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []byte, maxPkt int, startAck Value) {
-	err := rtx.Reset(buf, maxPkt, startAck)
-	if err != nil {
-		t.Fatal(err)
-	}
-	prevSeq := Value(startAck)
-	for i, msg := range msgs {
-		if t.Failed() {
-			t.Errorf("%s failed on message %d", t.Name(), i)
-			return
-		}
-		if len(aux) < len(msg) {
-			panic("need aux to contain message")
-		}
-		n, err := rtx.Write(msg)
-		if err != nil {
-			t.Fatalf("writing packet %d: %s", i, err)
-		} else if n != len(msg) {
-			t.Fatalf("want %d written, got %d", len(msg), n)
-		}
-		testQueueSanity(t, rtx)
-		unsent := rtx.Buffered()
-		if len(msg) != unsent {
-			t.Fatalf("want %d unsent buffered, got %d", len(msg), unsent)
-		}
-		testQueueSanity(t, rtx)
-		sent := rtx.BufferedSent()
-		if sent != 0 {
-			t.Fatalf("want 0 bytes sent, got %d", sent)
-		}
-		testQueueSanity(t, rtx)
-		n, seq, err := rtx.MakePacket(aux[:])
-		data := aux[:n]
-		if err != nil {
-			t.Fatalf("making packet %d: %s", i, err)
-		} else if n != len(msg) {
-			t.Fatalf("want %d packet read, got %d", len(msg), n)
-		} else if !bytes.Equal(msg, aux[:n]) {
-			t.Fatalf("want data %q, got data read %q", msg, data[:n])
-		} else if seq != prevSeq {
-			t.Fatalf("want seq %d, got %d", prevSeq, seq)
-		}
-		testQueueSanity(t, rtx)
-		sent = rtx.BufferedSent()
-		if sent != len(msg) {
-			t.Fatalf("want %d sent, got %d", len(msg), sent)
-		}
-		testQueueSanity(t, rtx)
-		prevSeq += Value(n)
-		err = rtx.RecvACK(prevSeq)
-		if err != nil {
-			t.Fatal(err)
-		}
-		sent = rtx.BufferedSent()
-		unsent = rtx.Buffered()
-		if sent != 0 {
-			t.Errorf("message not marked as sent- expected no data left got %d", sent)
-		}
-		if unsent != 0 {
-			t.Errorf("huge bug, unexpected data loaded to unsent buffer")
-		}
-		testQueueSanity(t, rtx)
-	}
-}
-
 func testQueueSanity(t *testing.T, rtx *ringTx) {
 	// t.Helper()
 	alreadyFailed := t.Failed()