feat(ntp): add two-exchange client, server handler, and extension fields (#101)

Implement full two-exchange NTP client state machine (RFC 5905 §8). First exchange stores offset/RTT, second averages both for improved accuracy. Client places T1 in TransmitTime per RFC 5905 §8; server echoes it as OriginTime in the response. Add NTP extension field codec (RFC 7822) with NextExtField iterator and AppendExtField builder. Add NTS extension type constants from RFC 8915. Add NTP Server (StackNode) that receives client requests via Demux and builds server responses via Encapsulate with configurable stratum, precision, reference ID, and pending request queue. Add Frame accessor methods: RawData(), ExtensionFields(), ValidateSize(), and Timestamp.Uint64(). Add ntp-client and ntp-server example programs with CalculateSystemPrecision, retry limits, and backoff. Use internal.LogAttrs pattern for non-allocating structured logging in the client, matching the tcp/debug.go convention. Generated with Claude assistance. Signed-off-by: Marvin Drees <marvin.drees@9elements.com>
2026-08-20 22:49:11 +00:00 · 2026-04-27 18:27:08 +02:00
parent cf94767133
commit bba913751a
10 changed files with 1030 additions and 68 deletions
@@ -1,9 +1,11 @@
 package ntp

 import (
+	"log/slog"
 	"time"

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

 type state uint8
@@ -23,14 +25,15 @@ type Client struct {
 	connID uint64
 	start  time.Time
 	_now   func() time.Time
-	// t stores the time offsets needed to compute the time at client
-	// taking into consideration the round-trip delay.
-	//  - t[0] (orig): Client timestamp of request packet transmission.
-	//  - t[1] (rec): Server timestamp of request packet reception.
-	//  - t[2] (xmt): Server timestamp of response packet transmission.
-	//  - t[3]: Client timestamp of response packet reception.
-	t [4]Timestamp
-	// org      Timestamp
+	logger logger
+	// t stores the four NTP timestamps per RFC 5905:
+	//  - t[0] (T1): Client timestamp of request packet transmission.
+	//  - t[1] (T2): Server timestamp of request packet reception.
+	//  - t[2] (T3): Server timestamp of response packet transmission.
+	//  - t[3] (T4): Client timestamp of response packet reception.
+	t             [4]Timestamp
+	offset1       time.Duration // clock offset from first exchange, averaged with second in OffsetUnsynced.
+	rtt1          time.Duration // round-trip delay from first exchange, averaged with second in RoundTripDelay.
 	state         state
 	serverStratum Stratum
 	sysprec       int8
@@ -40,11 +43,16 @@ func (c *Client) Reset(sysprec int8, now func() time.Time) {
 	*c = Client{
 		connID:  c.connID + 1,
 		_now:    now,
+		logger:  c.logger,
 		sysprec: sysprec,
 		state:   stateSend1,
 	}
 }

+// SetLogger configures a structured logger for debug output.
+// Pass nil to disable logging (the default).
+func (c *Client) SetLogger(l *slog.Logger) { c.logger.log = l }
+
 func (c *Client) Protocol() uint64  { return 0 }
 func (c *Client) LocalPort() uint16 { return ClientPort }
 func (c *Client) ConnectionID() *uint64 {
@@ -62,27 +70,32 @@ func (c *Client) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int)
 	}

 	switch c.state {
-	case stateSend1:
-		c.start = c.now()
-		c.t[0] = TimestampFromUint64(0)
-		c.state = stateAwait1
-	case stateSend2:
-		// c.xmt = c.unsyncTimestamp(c.now())
-		c.state = stateDone
+	case stateSend1, stateSend2:
+		now := c.now()
+		c.start = now
+		var err error
+		if c.t[0], err = TimestampFromTime(now); err != nil {
+			return 0, err
+		}
+		if c.state == stateSend1 {
+			c.state = stateAwait1
+		} else {
+			c.state = stateAwait2
+		}
 	default:
 		return 0, nil // Nothing to handle.
 	}

-	for i := range payload[:SizeHeader] {
-		payload[i] = 0
-	}
-
 	frm.ClearHeader()
 	frm.SetStratum(StratumUnsync)
 	frm.SetPoll(6)
 	frm.SetPrecision(c.sysprec)
-	frm.SetOriginTime(c.t[0])
+	// RFC 5905 §8: client places T1 in TransmitTime of the request.
+	// The server will echo it back as OriginTime in its response.
+	frm.SetTransmitTime(c.t[0])
 	frm.SetFlags(ModeClient, Version4, LeapNoWarning)
+	c.logger.debug("ntp.Client:encapsulate", slog.Int("state", int(c.state)),
+		slog.Uint64("T1", c.t[0].Uint64()))
 	return SizeHeader, nil
 }

@@ -97,21 +110,45 @@ func (c *Client) Demux(carrierData []byte, frameOffset int) error {
 	}

 	switch c.state {
-	case stateAwait1:
-		xmt := frm.TransmitTime()
-		orig := frm.OriginTime()
-		if xmt == orig || orig != c.t[0] {
-			return lneto.ErrPacketDrop
-		}
+	case stateAwait1, stateAwait2:
+	default:
+		return nil // Not awaiting a response.
+	}

-		txelapsed := c.now().Sub(c.start)
-		c.t[1] = frm.ReceiveTime()
-		c.t[2] = xmt
-		c.t[3] = c.t[0].Add(txelapsed)
+	// RFC 5905 §8 validation: discard bogus packets.
+	// Bogus: origin timestamp does not echo our T1 (the transmit time we sent).
+	// Malformed: server's transmit time equals its own origin echo.
+	xmt := frm.TransmitTime()
+	orig := frm.OriginTime()
+	if xmt == orig || orig != c.t[0] {
+		c.logger.debug("ntp.Client:demux:drop", slog.String("reason", "origin mismatch"),
+			slog.Uint64("orig", orig.Uint64()), slog.Uint64("T1", c.t[0].Uint64()))
+		return lneto.ErrPacketDrop
+	}
+
+	// Compute T4, then derive offset θ and round-trip delay δ per RFC 5905 §8.
+	txelapsed := c.now().Sub(c.start)
+	c.t[1] = frm.ReceiveTime()
+	c.t[2] = xmt
+	c.t[3] = c.t[0].Add(txelapsed)
+
+	offset := (c.t[1].Sub(c.t[0]) + c.t[2].Sub(c.t[3])) / 2
+	rtt := c.t[3].Sub(c.t[0]) - c.t[2].Sub(c.t[1])
+
+	if c.state == stateAwait1 {
 		c.serverStratum = frm.Stratum()
-		c.state = stateDone // TODO: add second exchange part.
-	case stateAwait2:
+		c.offset1 = offset
+		c.rtt1 = rtt
+		c.state = stateSend2
+		c.logger.debug("ntp.Client:demux:exchange1",
+			slog.Duration("offset", c.offset1), slog.Duration("rtt", c.rtt1),
+			slog.String("stratum", c.serverStratum.String()))
+	} else {
 		c.state = stateDone
+		c.logger.debug("ntp.Client:demux:exchange2",
+			slog.Duration("offset", offset), slog.Duration("rtt", rtt),
+			slog.Duration("avg_offset", (c.offset1+offset)/2),
+			slog.Duration("avg_rtt", (c.rtt1+rtt)/2))
 	}
 	return nil
 }
@@ -147,28 +184,35 @@ func (c *Client) Offset() time.Duration {
 }

 func (c *Client) offsetAndNow() (clientNow time.Time, offset time.Duration) {
-	now := c.now()
-	serverToBase := c.OffsetUnsynced()
-	clientToBase := now.Sub(BaseTime())
-	serverToClient := serverToBase - clientToBase
-	return now, serverToClient
+	return c.now(), c.OffsetUnsynced()
 }

 // OffsetUnsynced returns the absolute time offset difference between client and server clock
-// as calculated by the clock synchonization algorithm. It is unsynchonized- the result of OffsetUnsynced will not change with time.
+// as calculated by the clock synchronization algorithm. It is unsynced — the result will not
+// change with time. When both exchanges are complete the result is the average of both exchanges.
 func (c *Client) OffsetUnsynced() time.Duration {
 	if c.IsDone() {
 		t := &c.t
-		return (t[1].Sub(t[0]) + t[2].Sub(t[3])) / 2
+		offset2 := (t[1].Sub(t[0]) + t[2].Sub(t[3])) / 2
+		return (c.offset1 + offset2) / 2
 	}
 	return 0
 }

+// RoundTripDelay returns the average round-trip delay across both NTP exchanges.
 func (c *Client) RoundTripDelay() time.Duration {
 	if c.IsDone() {
-		d0 := c.t[3].Sub(c.t[0])
-		d1 := c.t[2].Sub(c.t[1])
-		return d0 - d1
+		rtt2 := c.t[3].Sub(c.t[0]) - c.t[2].Sub(c.t[1])
+		return (c.rtt1 + rtt2) / 2
 	}
 	return -1
 }
+
+// logger provides non-allocating structured logging using [internal.LogAttrs].
+type logger struct {
+	log *slog.Logger
+}
+
+func (l logger) debug(msg string, attrs ...slog.Attr) {
+	internal.LogAttrs(l.log, slog.LevelDebug, msg, attrs...)
+}