From 64a710206482bbdb20d94e1d8fa64ea2ab82d09b Mon Sep 17 00:00:00 2001 From: soypat Date: Sun, 9 Feb 2025 23:43:30 -0300 Subject: [PATCH] continue adding Handler logic; remove unused TxQueue tests --- internal/debug_heaplog.go | 4 + internal/debug_noheaplog.go | 4 + lneto2/crc.go | 84 +++++++++++++++ lneto2/validation.go | 64 ++++++++++++ tcp/control.go | 4 +- tcp/debug.go | 32 +++--- tcp/frame.go | 198 ++++++++++++++++++++++++++++++++++++ tcp/handler.go | 132 ++++++++++++++++++++++++ tcp/txqueue_test.go | 130 +++-------------------- 9 files changed, 520 insertions(+), 132 deletions(-) create mode 100644 lneto2/crc.go create mode 100644 lneto2/validation.go create mode 100644 tcp/frame.go create mode 100644 tcp/handler.go diff --git a/internal/debug_heaplog.go b/internal/debug_heaplog.go index 4fb70f9..c300d9e 100644 --- a/internal/debug_heaplog.go +++ b/internal/debug_heaplog.go @@ -21,6 +21,10 @@ var ( timebuf [len(timefmt) * 2]byte ) +func LogEnabled(l *slog.Logger, lvl slog.Level) bool { + return true +} + func LogAttrs(_ *slog.Logger, level slog.Level, msg string, attrs ...slog.Attr) { now := time.Now() n := len(now.AppendFormat(timebuf[:0], timefmt)) diff --git a/internal/debug_noheaplog.go b/internal/debug_noheaplog.go index 4b4f3b2..a39a0be 100644 --- a/internal/debug_noheaplog.go +++ b/internal/debug_noheaplog.go @@ -9,6 +9,10 @@ import ( const HeapAllocDebugging = false +func LogEnabled(l *slog.Logger, lvl slog.Level) bool { + return l != nil && l.Handler().Enabled(context.Background(), lvl) +} + // LogAttrs is a helper function that is used by all package loggers and that // can be switched out with the `debugheaplog` build tag for a non-allocating // logger that prints out when heap allocations occur. diff --git a/lneto2/crc.go b/lneto2/crc.go new file mode 100644 index 0000000..8e71af4 --- /dev/null +++ b/lneto2/crc.go @@ -0,0 +1,84 @@ +package lneto2 + +import ( + "encoding/binary" +) + +// CRC791 function as defined by RFC 791. The Checksum field for TCP+IP +// is the 16-bit ones' complement of the ones' complement sum of +// all 16-bit words in the header. In case of uneven number of octet the +// last word is LSB padded with zeros. +// +// The zero value of CRC791 is ready to use. +type CRC791 struct { + sum uint32 + excedent uint8 + needPad bool +} + +// Write adds the bytes in p to the running checksum. +func (c *CRC791) Write(buff []byte) (n int, err error) { + if len(buff) == 0 { + return 0, nil + } + if c.needPad { + c.sum += uint32(c.excedent)<<8 + uint32(buff[0]) + buff = buff[1:] + c.excedent = 0 + c.needPad = false + if len(buff) == 0 { + return 1, nil + } + } + count := len(buff) + for count > 1 { + c.sum += uint32(binary.BigEndian.Uint16(buff[len(buff)-count:])) + count -= 2 + } + if count != 0 { + c.excedent = buff[len(buff)-1] + c.needPad = true + } + return len(buff), nil +} + +// AddUint32 adds a 32 bit value to the running checksum interpreted as BigEndian (network order). +func (c *CRC791) AddUint32(value uint32) { + c.AddUint16(uint16(value >> 16)) + c.AddUint16(uint16(value)) +} + +// Add16 adds a 16 bit value to the running checksum interpreted as BigEndian (network order). +func (c *CRC791) AddUint16(value uint16) { + if c.needPad { + c.sum += uint32(c.excedent)<<8 | uint32(value>>8) + c.excedent = byte(value) + } else { + c.sum += uint32(value) + } +} + +// Add16 adds value to the running checksum interpreted as BigEndian (network order). +func (c *CRC791) AddUint8(value uint8) { + if c.needPad { + c.sum += uint32(c.excedent)<<8 | uint32(value) + } else { + c.excedent = value + } + c.needPad = !c.needPad +} + +// Sum16 calculates the checksum with the data written to c thus far. +func (c *CRC791) Sum16() uint16 { + sum := c.sum + if c.needPad { + sum += uint32(c.excedent) << 8 + } + for sum>>16 != 0 { + sum = (sum & 0xffff) + (sum >> 16) + } + return uint16(^sum) +} + +// Reset zeros out the CRC791, resetting it to the initial state. +func (c *CRC791) Reset() { *c = CRC791{} } diff --git a/lneto2/validation.go b/lneto2/validation.go new file mode 100644 index 0000000..34baa98 --- /dev/null +++ b/lneto2/validation.go @@ -0,0 +1,64 @@ +package lneto2 + +import ( + "errors" + "fmt" +) + +type Validator struct { + checkEvil bool + allowMultiErrs bool + accum []error + accumBitpos []BitPosErr +} + +func (v *Validator) ResetErr() { + v.accum = v.accum[:0] + v.accumBitpos = v.accumBitpos[:0] +} + +func (v *Validator) HasError() bool { + return len(v.accum) != 0 +} + +func (v *Validator) Err() error { + if len(v.accum) == 1 { + return v.accum[0] + } else if len(v.accum) == 0 { + return nil + } + return errors.Join(v.accum...) +} + +func (v *Validator) gotErr(err error) { + v.accum = append(v.accum, err) +} + +func (v *Validator) AddError(err error) { + if err == nil { + panic("error argument to AddError cannot be nil") + } else if len(v.accum) != 0 && !v.allowMultiErrs { + return + } + v.accum = append(v.accum, err) +} + +func (v *Validator) AddBitPosErr(bitStart, bitLen int, err error) { + if err == nil { + panic("err argument to bitPosErr cannot be nil") + } else if bitLen <= 0 { + panic("") + } + v.accumBitpos = append(v.accumBitpos, BitPosErr{BitStart: bitStart, BitLen: bitLen, Err: err}) + v.accum = append(v.accum, &v.accumBitpos[len(v.accumBitpos)-1]) +} + +type BitPosErr struct { + BitStart int + BitLen int + Err error +} + +func (bpe *BitPosErr) Error() string { + return fmt.Sprintf("%s at bits %d..%d", bpe.Err.Error(), bpe.BitStart, bpe.BitStart+bpe.BitLen) +} diff --git a/tcp/control.go b/tcp/control.go index 4c6bf8b..9c45b2f 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -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. diff --git a/tcp/debug.go b/tcp/debug.go index d8f6462..5bb109b 100644 --- a/tcp/debug.go +++ b/tcp/debug.go @@ -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) { diff --git a/tcp/frame.go b/tcp/frame.go new file mode 100644 index 0000000..78f3d59 --- /dev/null +++ b/tcp/frame.go @@ -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) + } +} diff --git a/tcp/handler.go b/tcp/handler.go new file mode 100644 index 0000000..90f363f --- /dev/null +++ b/tcp/handler.go @@ -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() +} diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index e5b471c..3be716b 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -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[:], ¤tAck) + 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()