server: upstream trains, observed TTL/DSCP/ECN, rate limits, action ids

Types 0x03/0x04/0x05 land with a bounded columnar train buffer (head kept,
truncation declared) and grant-free multi-part reports - a report row is
smaller than the packet it answers, so $3.4 holds without a grant. The
read loop now collects TTL/TOS cmsgs on Linux, replacing the 0xFF stubs in
the observation block with what the kernel saw; downtrain gained a dscp
parameter, so DSCP survival is measurable in both directions.

Rate limiting ($2.5) exists now: per-credential AND per-source buckets,
429 on the control plane, silent drop on the data plane after the HMAC
gate and before the replay window. UDP ceilings default above the largest
legitimate run - a limit that clips a real measurement produces a
confidently wrong number.

Every granted packet carries its action_id at payload[8:16]; overlapping
actions were unattributable before. Canary DNS logs now honor the stated
24h privacy default. /admin/enroll-tokens answers the spec's JSON shape.
protocol_version 1.0.1 (additive).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

server/internal/dataplane/cmsg_linux.go

@@ -0,0 +1,70 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+//go:build linux
+
+package dataplane
+
+import (
+	"encoding/binary"
+	"net"
+	"syscall"
+)
+
+// Per-packet TTL and TOS/traffic-class arrive as control messages, and only if asked for at
+// socket setup. These fill the spec §3.3 observation-block fields that were shipped as the 0xFF
+// sentinel until now — received TTL is path-length evidence, the TOS byte is DSCP/ECN survival.
+
+// enableRecvMeta asks the kernel to attach the cmsgs to every received datagram. Both the v4 and
+// the v6 option sets are attempted on every socket: a dual-stack socket delivers v4-mapped
+// traffic through the v6 fd, and the kernel refuses whichever set does not apply. Errors are
+// dropped on purpose — a socket that cannot deliver metadata still serves probes, and the
+// sentinel already says "not observed" for it.
+func enableRecvMeta(conn *net.UDPConn) {
+	raw, err := conn.SyscallConn()
+	if err != nil {
+		return
+	}
+	_ = raw.Control(func(fd uintptr) {
+		_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_RECVTTL, 1)
+		_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_RECVTOS, 1)
+		_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_RECVHOPLIMIT, 1)
+		_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_RECVTCLASS, 1)
+	})
+}
+
+// parseMeta extracts TTL and the TOS byte from one datagram's control messages. Anything absent
+// or unparseable keeps the sentinel — reported as unobserved, never guessed.
+func parseMeta(oob []byte) pktMeta {
+	m := pktMeta{TTL: metaUnavailable, TOS: metaUnavailable}
+	if len(oob) == 0 {
+		return m
+	}
+	cmsgs, err := syscall.ParseSocketControlMessage(oob)
+	if err != nil {
+		return m
+	}
+	for _, c := range cmsgs {
+		switch {
+		case c.Header.Level == syscall.IPPROTO_IP && c.Header.Type == syscall.IP_TTL,
+			c.Header.Level == syscall.IPPROTO_IPV6 && c.Header.Type == syscall.IPV6_HOPLIMIT:
+			m.TTL = cmsgValue(c.Data)
+		case c.Header.Level == syscall.IPPROTO_IP && c.Header.Type == syscall.IP_TOS,
+			c.Header.Level == syscall.IPPROTO_IPV6 && c.Header.Type == syscall.IPV6_TCLASS:
+			m.TOS = cmsgValue(c.Data)
+		}
+	}
+	return m
+}
+
+// cmsgValue reads a cmsg the kernel encodes either as a native-endian int (IP_TTL,
+// IPV6_HOPLIMIT, IPV6_TCLASS) or as a single byte (IP_TOS). Both fit a byte by definition.
+func cmsgValue(data []byte) uint8 {
+	switch {
+	case len(data) >= 4:
+		return uint8(binary.NativeEndian.Uint32(data))
+	case len(data) >= 1:
+		return data[0]
+	}
+	return metaUnavailable
+}

server/internal/dataplane/cmsg_other.go

@@ -0,0 +1,15 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+//go:build !linux
+
+package dataplane
+
+import "net"
+
+// Per-packet TTL/TOS needs Linux's IP_RECVTTL-family cmsgs. Elsewhere the observation block and
+// train buffers keep the spec §3.3 sentinel (0xFF = not observed) — absent is honest, a guess
+// is not. Deployment targets are Linux; this build exists so the Windows dev loop compiles.
+func enableRecvMeta(_ *net.UDPConn) {}
+
+func parseMeta(_ []byte) pktMeta { return pktMeta{TTL: metaUnavailable, TOS: metaUnavailable} }

server/internal/dataplane/frag_linux.go

@@ -104,8 +104,8 @@ func (s *Server) FragSend(
 		return res, fmt.Errorf("session has no recorded local address")
 	}
 
-	if sizeBytes < HeaderSize+8 {
-		sizeBytes = HeaderSize + 8
+	if sizeBytes < HeaderSize+32 {
+		sizeBytes = HeaderSize + 32
 	}
 	if sizeBytes > 8000 {
 		sizeBytes = 8000
@@ -117,7 +117,10 @@ func (s *Server) FragSend(
 	// The ELT1 packet, signed exactly as any other, then wrapped in UDP.
 	payload := make([]byte, sizeBytes-HeaderSize)
 	binary.BigEndian.PutUint32(payload[0:4], uint32(sizeBytes))
-	copy(payload[4:], mode)
+	// [8:16]: the action id, as on every granted packet (spec §5 correlation); the mode string
+	// sits past the reserved slot.
+	putActionID(payload, g.ActionID)
+	copy(payload[16:], mode)
 	elt := s.buildPacket(sess, TypeFragData, 0, payload)
 
 	udp := buildUDP(local, target, elt)

server/internal/dataplane/granted.go

@@ -21,7 +21,11 @@ import (
 // client measures downstream loss, reordering and jitter from what arrives — the direction an
 // upstream-only train cannot see. Returns how many packets actually went out (the grant may cut
 // it short, which is itself reportable).
-func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error) {
+//
+// dscp ≥ 0 marks the burst (spec §5 downtrain `dscp`): downstream DSCP survival is the half the
+// client cannot produce itself. Best-effort off Linux — see withTOS/TOSSupported; the action
+// response has already told the client whether the marking was applied.
+func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs, dscp int) (int, error) {
 	target := sess.DataSource()
 	if !target.IsValid() {
 		return 0, fmt.Errorf("no observed data-plane source")
@@ -30,26 +34,41 @@ func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeB
 	if conn == nil {
 		return 0, fmt.Errorf("no data-plane socket matches target family")
 	}
-	if sizeBytes < HeaderSize+8 {
-		sizeBytes = HeaderSize + 8
+	if sizeBytes < HeaderSize+16 {
+		sizeBytes = HeaderSize + 16
 	}
 	payload := make([]byte, sizeBytes-HeaderSize)
+	// Payload [8:16] carries the action id on every granted packet, so arriving traffic can be
+	// attributed to the action that caused it (spec §5/§9: test.params.action_id).
+	putActionID(payload, g.ActionID)
 	sent := 0
-	for i := 0; i < count; i++ {
-		if !g.Allow(sizeBytes) {
-			break // budget or rate exhausted — stop, do not sleep it off
-		}
-		// Sequence + send timestamp in the payload head so the client can order and time them
-		// even when packets arrive out of order.
-		binary.BigEndian.PutUint32(payload[0:4], uint32(i))
-		binary.BigEndian.PutUint32(payload[4:8], uint32(time.Since(s.start).Microseconds()))
-		s.send(conn, target, sess, TypeDownTrainData, uint32(i), payload)
-		sent++
-		if intervalUs > 0 && i < count-1 {
-			time.Sleep(time.Duration(intervalUs) * time.Microsecond)
+	burst := func() error {
+		for i := 0; i < count; i++ {
+			if !g.Allow(sizeBytes) {
+				break // budget or rate exhausted — stop, do not sleep it off
+			}
+			// Sequence + send timestamp in the payload head so the client can order and time them
+			// even when packets arrive out of order.
+			binary.BigEndian.PutUint32(payload[0:4], uint32(i))
+			binary.BigEndian.PutUint32(payload[4:8], uint32(time.Since(s.start).Microseconds()))
+			s.send(conn, target, sess, TypeDownTrainData, uint32(i), payload)
+			sent++
+			if intervalUs > 0 && i < count-1 {
+				time.Sleep(time.Duration(intervalUs) * time.Microsecond)
+			}
 		}
+		return nil
 	}
-	return sent, nil
+	if dscp >= 0 && TOSSupported {
+		// Same shared-socket borrow as the DF window: dfMu keeps a concurrent burst from riding
+		// along with — or clearing — this marking.
+		s.dfMu.Lock()
+		defer s.dfMu.Unlock()
+		err := withTOS(conn, dscp, burst) // sent must be read after the burst ran, not before
+		return sent, err
+	}
+	err := burst()
+	return sent, err
 }
 
 // BigSendResult records what happened to one requested size. `Sent` false with an EMSGSIZE-ish
@@ -83,8 +102,8 @@ func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, d
 	results := make([]BigSendResult, 0, len(sizes))
 	burst := func() error {
 		for i, size := range sizes {
-			if size < HeaderSize+8 {
-				size = HeaderSize + 8
+			if size < HeaderSize+16 {
+				size = HeaderSize + 16
 			}
 			if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway
 				size = 9000
@@ -96,6 +115,8 @@ func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, d
 			// Echo the intended size into the payload so a truncated/fragmented arrival is
 			// still attributable to the size we meant to send.
 			binary.BigEndian.PutUint32(payload[0:4], uint32(size))
+			// [8:16]: the action id, as on every granted packet (spec §5 correlation).
+			putActionID(payload, g.ActionID)
 			err := s.sendErr(conn, target, sess, TypeBigSend, uint32(i), payload)
 			results = append(results, BigSendResult{
 				SizeBytes: size, Seq: i, Sent: err == nil, Err: errString(err),

server/internal/dataplane/throughput.go

@@ -120,7 +120,7 @@ func (s *Server) DownThroughput(
 
 	// Same plan the grant was sized from, so the two cannot disagree.
 	durationMs, kbps = ThroughputPlan(durationMs, kbps)
-	if sizeBytes < HeaderSize+16 {
+	if sizeBytes < HeaderSize+24 {
 		sizeBytes = 1200 // a size that survives every common path unfragmented
 	}
 	if sizeBytes > 1472 {
@@ -134,6 +134,10 @@ func (s *Server) DownThroughput(
 	}
 
 	payload := make([]byte, sizeBytes-HeaderSize)
+	// [8:16]: the action id, as on every granted packet (spec §5 correlation). The send
+	// timestamp lives past it at [16:24]; the client reads only header fields today, so
+	// reserving the slot costs nothing and keeps one layout rule across granted types.
+	putActionID(payload, g.ActionID)
 	deadline := time.Now().Add(time.Duration(durationMs) * time.Millisecond)
 	start := time.Now()
 	next := start
@@ -157,7 +161,7 @@ func (s *Server) DownThroughput(
 		// Reaching here means the run is progressing normally; the clock will end it.
 		res.LimitedBy = "duration"
 		binary.BigEndian.PutUint32(payload[0:4], seq)
-		binary.BigEndian.PutUint64(payload[4:12], uint64(time.Since(s.start).Nanoseconds()))
+		binary.BigEndian.PutUint64(payload[16:24], uint64(time.Since(s.start).Nanoseconds()))
 		if err := s.sendErr(conn, target, sess, TypeThroughputData, seq, payload); err != nil {
 			// A send error mid-run is a local condition (buffer full, route gone). Stop and
 			// report what got out rather than pretending the rest was lost on the path.

server/internal/dataplane/tos_linux.go

@@ -0,0 +1,55 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+//go:build linux
+
+package dataplane
+
+import (
+	"net"
+	"syscall"
+)
+
+// withTOS runs fn with the socket's TOS/traffic class set to dscp<<2 (ECN bits left zero — the
+// test is about DSCP survival, and claiming ECN capability we do not use would pollute it), then
+// restores what was there before.
+//
+// Same borrow discipline as withDF: the socket is shared by every session on that family, so the
+// caller must hold Server.dfMu for the whole window or a concurrent burst rides along with — or
+// clears — someone else's marking.
+func withTOS(conn *net.UDPConn, dscp int, fn func() error) error {
+	raw, err := conn.SyscallConn()
+	if err != nil {
+		return err
+	}
+	v4 := conn.LocalAddr().(*net.UDPAddr).IP.To4() != nil
+	level, opt := syscall.IPPROTO_IPV6, syscall.IPV6_TCLASS
+	if v4 {
+		level, opt = syscall.IPPROTO_IP, syscall.IP_TOS
+	}
+
+	var setErr error
+	prev := 0
+	if err := raw.Control(func(fd uintptr) {
+		if p, e := syscall.GetsockoptInt(int(fd), level, opt); e == nil {
+			prev = p
+		}
+		setErr = syscall.SetsockoptInt(int(fd), level, opt, dscp<<2)
+	}); err != nil {
+		return err
+	}
+	if setErr != nil {
+		return setErr
+	}
+	defer func() {
+		_ = raw.Control(func(fd uintptr) {
+			_ = syscall.SetsockoptInt(int(fd), level, opt, prev)
+		})
+	}()
+	return fn()
+}
+
+// TOSSupported reports whether withTOS can actually mark packets here. Exported so the control
+// plane can tell the client up front that its dscp request will not be honored, instead of the
+// client measuring an unmarked burst and concluding the network stripped the marking.
+const TOSSupported = true

server/internal/dataplane/tos_other.go

@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+//go:build !linux
+
+package dataplane
+
+import "net"
+
+// Setting DSCP per-burst uses IP_TOS/IPV6_TCLASS under the same fd-borrow pattern as withDF,
+// which is only exercised on Linux deployments. Elsewhere the burst goes out with the default
+// class and TOSSupported lets the action response say so — an unmarked burst reported as marked
+// would read as "the network stripped DSCP", the exact wrong conclusion.
+func withTOS(_ *net.UDPConn, _ int, fn func() error) error { return fn() }
+
+const TOSSupported = false

server/internal/dataplane/train.go

@@ -0,0 +1,122 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+// Upstream trains (spec §3.2, types 0x03–0x05). The client blasts TRAIN_DATA at the server and
+// the server answers nothing per-packet — a reply would double the traffic and measure the
+// return path at the same time. Afterwards the client asks for the server's received view with
+// TRAIN_REPORT_REQ, and gets it back columnar, split across as many TRAIN_REPORT datagrams as
+// it takes to stay under a safe size.
+//
+// Both TRAIN_DATA and TRAIN_REPORT_REQ carry the train id in payload[0:4]; the id is the
+// client's to choose, unique within the session.
+
+import (
+	"encoding/binary"
+	"net"
+	"net/netip"
+
+	"echo-lot.app/server/internal/session"
+)
+
+const (
+	// trainReportMaxDatagram caps one TRAIN_REPORT datagram at a size that survives every common
+	// path unfragmented. A report about loss must not itself be lost to MTU.
+	trainReportMaxDatagram = 1200
+	trainReportHeader      = 16
+	trainReportRow         = 17 // 4 seq + 8 t_rx_ns + 2 size + 1 ttl + 1 dscp + 1 ecn
+)
+
+// recordTrain buffers one TRAIN_DATA packet into its train (session-side, bounded — see
+// session/train.go). A payload too short to carry the id is unreportable and stays only in the
+// flat packet log, which already recorded it.
+func recordTrain(sess *session.Session, payload []byte, seq uint32, size int, tRxNs int64, meta pktMeta) {
+	if len(payload) < 4 {
+		return
+	}
+	sess.RecordTrainPacket(binary.BigEndian.Uint32(payload[0:4]), session.TrainEntry{
+		Seq: seq, TRxNs: tRxNs, Size: uint16(min(size, 0xFFFF)),
+		TTL: meta.TTL, DSCP: meta.dscp(), ECN: meta.ecn(),
+	})
+}
+
+// trainReport answers one TRAIN_REPORT_REQ with the full columnar report.
+//
+// Grant-free on purpose. §3.4 caps ungranted responses at the request size, and a multi-part
+// report is larger than the single REPORT_REQ that asked for it — but it cannot amplify: every
+// 17-byte row accounts for one HMAC-valid TRAIN_DATA packet of at least HeaderSize+4 bytes this
+// session already delivered here, so the whole report is a strict fraction of the traffic it
+// describes, and it only ever goes to the session's verified source address. An unknown id gets
+// a single zero-row report rather than silence — "nothing arrived" IS the measurement.
+func (s *Server) trainReport(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, payload []byte) {
+	if len(payload) < 4 {
+		return
+	}
+	id := binary.BigEndian.Uint32(payload[0:4])
+	train, _ := sess.TrainView(id)
+	train.ID = id
+	for i, part := range buildTrainReport(train) {
+		s.send(conn, raddr, sess, TypeTrainReport, uint32(i), part)
+	}
+}
+
+// buildTrainReport lays a train's received view out columnar and cuts it into datagram-sized
+// payloads. Layout (mirrored by the client; all big-endian):
+//
+//	 0  4  train_id
+//	 4  4  received       (every packet counted, buffered or not — loss math uses this)
+//	 8  2  part           (0-based)
+//	10  2  parts
+//	12  1  flags          (bit0: buffer overflowed; rows beyond the cap were counted, not kept —
+//	                       the schema's evidence_truncated honesty, on the wire)
+//	13  1  reserved
+//	14  2  n              (rows in this part)
+//	16     n×4 seq, n×8 t_rx_ns, n×2 size, n×1 ttl, n×1 dscp, n×1 ecn (columns contiguous)
+func buildTrainReport(t session.Train) [][]byte {
+	perPart := (trainReportMaxDatagram - HeaderSize - trainReportHeader) / trainReportRow
+	parts := (len(t.Entries) + perPart - 1) / perPart
+	if parts == 0 {
+		parts = 1 // an empty train still gets its "received: 0" answer
+	}
+	out := make([][]byte, 0, parts)
+	for p := 0; p < parts; p++ {
+		rows := t.Entries[p*perPart : min((p+1)*perPart, len(t.Entries))]
+		n := len(rows)
+		b := make([]byte, trainReportHeader+n*trainReportRow)
+		binary.BigEndian.PutUint32(b[0:4], t.ID)
+		binary.BigEndian.PutUint32(b[4:8], uint32(t.Received))
+		binary.BigEndian.PutUint16(b[8:10], uint16(p))
+		binary.BigEndian.PutUint16(b[10:12], uint16(parts))
+		if t.Truncated {
+			b[12] = 1
+		}
+		binary.BigEndian.PutUint16(b[14:16], uint16(n))
+		off := trainReportHeader
+		for i, r := range rows {
+			binary.BigEndian.PutUint32(b[off+i*4:], r.Seq)
+		}
+		off += n * 4
+		for i, r := range rows {
+			binary.BigEndian.PutUint64(b[off+i*8:], uint64(r.TRxNs))
+		}
+		off += n * 8
+		for i, r := range rows {
+			binary.BigEndian.PutUint16(b[off+i*2:], r.Size)
+		}
+		off += n * 2
+		for i, r := range rows {
+			b[off+i] = r.TTL
+		}
+		off += n
+		for i, r := range rows {
+			b[off+i] = r.DSCP
+		}
+		off += n
+		for i, r := range rows {
+			b[off+i] = r.ECN
+		}
+		out = append(out, b)
+	}
+	return out
+}

server/internal/dataplane/train_test.go

@@ -0,0 +1,179 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+import (
+	"encoding/binary"
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+
+	"echo-lot.app/server/internal/session"
+)
+
+// parseReportPart decodes one TRAIN_REPORT payload back into rows, checking the header.
+func parseReportPart(t *testing.T, b []byte) (id uint32, received int, part, parts int, truncated bool, rows []session.TrainEntry) {
+	t.Helper()
+	if len(b) < trainReportHeader {
+		t.Fatalf("report part shorter than its header: %d", len(b))
+	}
+	id = binary.BigEndian.Uint32(b[0:4])
+	received = int(binary.BigEndian.Uint32(b[4:8]))
+	part = int(binary.BigEndian.Uint16(b[8:10]))
+	parts = int(binary.BigEndian.Uint16(b[10:12]))
+	truncated = b[12]&1 != 0
+	n := int(binary.BigEndian.Uint16(b[14:16]))
+	if want := trainReportHeader + n*trainReportRow; len(b) != want {
+		t.Fatalf("part length %d, want %d for %d rows", len(b), want, n)
+	}
+	off := trainReportHeader
+	rows = make([]session.TrainEntry, n)
+	for i := range rows {
+		rows[i].Seq = binary.BigEndian.Uint32(b[off+i*4:])
+	}
+	off += n * 4
+	for i := range rows {
+		rows[i].TRxNs = int64(binary.BigEndian.Uint64(b[off+i*8:]))
+	}
+	off += n * 8
+	for i := range rows {
+		rows[i].Size = binary.BigEndian.Uint16(b[off+i*2:])
+	}
+	off += n * 2
+	for i := range rows {
+		rows[i].TTL = b[off+i]
+	}
+	off += n
+	for i := range rows {
+		rows[i].DSCP = b[off+i]
+	}
+	off += n
+	for i := range rows {
+		rows[i].ECN = b[off+i]
+	}
+	return
+}
+
+func TestBuildTrainReportSplitsAndRoundTrips(t *testing.T) {
+	const count = 250 // enough to need several parts
+	train := session.Train{ID: 42, Received: count, Truncated: true}
+	for i := 0; i < count; i++ {
+		train.Entries = append(train.Entries, session.TrainEntry{
+			Seq: uint32(i), TRxNs: int64(i) * 1_000_000, Size: uint16(100 + i),
+			TTL: 64, DSCP: 46, ECN: 1,
+		})
+	}
+
+	parts := buildTrainReport(train)
+	if len(parts) < 2 {
+		t.Fatalf("250 rows should not fit one ≤%d-byte datagram", trainReportMaxDatagram)
+	}
+	var got []session.TrainEntry
+	for i, p := range parts {
+		if HeaderSize+len(p) > trainReportMaxDatagram {
+			t.Fatalf("part %d would be a %d-byte datagram, cap is %d", i, HeaderSize+len(p), trainReportMaxDatagram)
+		}
+		id, received, part, total, truncated, rows := parseReportPart(t, p)
+		if id != 42 || received != count || part != i || total != len(parts) || !truncated {
+			t.Fatalf("part %d header: id=%d received=%d part=%d/%d truncated=%v",
+				i, id, received, part, total, truncated)
+		}
+		got = append(got, rows...)
+	}
+	if len(got) != count {
+		t.Fatalf("round-tripped %d rows, want %d", len(got), count)
+	}
+	for i, r := range got {
+		want := train.Entries[i]
+		if r != want {
+			t.Fatalf("row %d = %+v, want %+v", i, r, want)
+		}
+	}
+}
+
+func TestBuildTrainReportEmptyTrainStillAnswers(t *testing.T) {
+	parts := buildTrainReport(session.Train{ID: 9})
+	if len(parts) != 1 {
+		t.Fatalf("empty train: %d parts, want 1 — 'nothing arrived' is the answer, not silence", len(parts))
+	}
+	id, received, _, total, _, rows := parseReportPart(t, parts[0])
+	if id != 9 || received != 0 || total != 1 || len(rows) != 0 {
+		t.Fatalf("empty report: id=%d received=%d parts=%d rows=%d", id, received, total, len(rows))
+	}
+}
+
+func TestTrainDataThenReportOverTheWire(t *testing.T) {
+	mgr, addr := startServer(t)
+	sess, _, err := mgr.New("dev1", "credential-ikm", netip.MustParseAddr("127.0.0.1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	client, err := net.DialUDP("udp", nil, net.UDPAddrFromAddrPort(addr))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	client.SetDeadline(time.Now().Add(3 * time.Second))
+
+	// A short train: id 5 in payload[0:4], plus padding.
+	const trainID, count = 5, 4
+	for i := 0; i < count; i++ {
+		payload := make([]byte, 60)
+		binary.BigEndian.PutUint32(payload[0:4], trainID)
+		if _, err := client.Write(craft(t, sess, TypeTrainData, uint32(i+1), payload)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// TRAIN_DATA must be silent (spec §3.2).
+	client.SetReadDeadline(time.Now().Add(300 * time.Millisecond))
+	if _, err := client.Read(make([]byte, 1500)); err == nil {
+		t.Fatal("TRAIN_DATA got a response, want none")
+	}
+
+	// Ask for the report.
+	reqPayload := make([]byte, 4)
+	binary.BigEndian.PutUint32(reqPayload, trainID)
+	client.SetDeadline(time.Now().Add(3 * time.Second))
+	if _, err := client.Write(craft(t, sess, TypeTrainReportReq, 100, reqPayload)); err != nil {
+		t.Fatal(err)
+	}
+	buf := make([]byte, 2000)
+	n, err := client.Read(buf)
+	if err != nil {
+		t.Fatalf("no TRAIN_REPORT: %v", err)
+	}
+	if buf[4] != TypeTrainReport {
+		t.Fatalf("type = %#x, want TRAIN_REPORT", buf[4])
+	}
+	id, received, part, parts, truncated, rows := parseReportPart(t, buf[HeaderSize:n])
+	if id != trainID || received != count || part != 0 || parts != 1 || truncated {
+		t.Fatalf("report header: id=%d received=%d part=%d/%d truncated=%v", id, received, part, parts, truncated)
+	}
+	if len(rows) != count {
+		t.Fatalf("%d rows, want %d", len(rows), count)
+	}
+	for i, r := range rows {
+		if r.Seq != uint32(i+1) {
+			t.Fatalf("row %d seq = %d, want %d", i, r.Seq, i+1)
+		}
+		if r.Size != HeaderSize+60 {
+			t.Fatalf("row %d size = %d, want %d", i, r.Size, HeaderSize+60)
+		}
+	}
+
+	// Unknown train id: one zero-row report, not silence.
+	binary.BigEndian.PutUint32(reqPayload, 999)
+	if _, err := client.Write(craft(t, sess, TypeTrainReportReq, 101, reqPayload)); err != nil {
+		t.Fatal(err)
+	}
+	n, err = client.Read(buf)
+	if err != nil {
+		t.Fatalf("no report for unknown train: %v", err)
+	}
+	id, received, _, _, _, rows = parseReportPart(t, buf[HeaderSize:n])
+	if id != 999 || received != 0 || len(rows) != 0 {
+		t.Fatalf("unknown-train report: id=%d received=%d rows=%d", id, received, len(rows))
+	}
+}

server/internal/dataplane/udp.go

@@ -2,15 +2,15 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 // Package dataplane implements the binary UDP probe protocol (spec §3):
-// 32-byte header, HMAC gate, anti-replay, ECHO with observation block.
-// Skeleton scope: ECHO_REQ/ECHO_RESP and TIMESYNC only; trains, MTU probes
-// and delayed echo land with the corresponding client tests.
+// 32-byte header, HMAC gate, anti-replay, ECHO with observation block,
+// upstream trains with columnar reports, and the granted server->client sends.
 package dataplane
 
 import (
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"log/slog"
 	"net"
@@ -18,6 +18,7 @@ import (
 	"sync"
 	"time"
 
+	"echo-lot.app/server/internal/ratelimit"
 	"echo-lot.app/server/internal/session"
 )
 
@@ -25,9 +26,14 @@ const (
 	Magic      = "ELT1"
 	HeaderSize = 32
 
-	TypeEchoReq     = 0x01
-	TypeEchoResp    = 0x02
-	TypeTimesyncReq = 0x07
+	TypeEchoReq  = 0x01
+	TypeEchoResp = 0x02
+	// Upstream trains (spec §3.2): DATA gets no per-packet response; REPORT_REQ fetches the
+	// server's received view as one or more REPORT datagrams (train.go).
+	TypeTrainData      = 0x03
+	TypeTrainReportReq = 0x04
+	TypeTrainReport    = 0x05
+	TypeTimesyncReq    = 0x07
 	TypeTimesyncRsp = 0x08
 	TypeMtuProbe    = 0x09
 	TypeMtuAck      = 0x0A
@@ -47,6 +53,12 @@ const (
 
 type Server struct {
 	Sessions *session.Manager
+	// Spec §2.5 ceilings on verified traffic, silent-drop (nil = no ceiling). Charged after the
+	// HMAC gate so an unauthenticated flood cannot spend anyone's budget, keyed per source
+	// address AND per device credential so neither one hot address nor one hot credential can
+	// crowd out the rest.
+	PacketRate *ratelimit.Limiter // tokens are packets
+	ByteRate   *ratelimit.Limiter // tokens are bytes
 	// Epoch for server-side t_rx/t_tx: process start; observation consumers
 	// only need differences plus the timesync exchange, not absolute time.
 	start time.Time
@@ -59,6 +71,34 @@ type Server struct {
 	dfMu sync.Mutex
 }
 
+// pktMeta is what the kernel told us about one received datagram beyond its bytes (spec §3.3:
+// received TTL, DSCP, ECN). 0xFF means "not observed": non-Linux hosts and datagrams whose
+// cmsg never arrived keep the sentinel rather than inventing a value.
+type pktMeta struct {
+	TTL uint8
+	TOS uint8 // the whole DSCP/ECN byte; DSCP = TOS>>2, ECN = TOS&3
+}
+
+const metaUnavailable = 0xFF
+
+func (m pktMeta) dscp() uint8 {
+	if m.TOS == metaUnavailable {
+		return metaUnavailable
+	}
+	return m.TOS >> 2
+}
+
+func (m pktMeta) ecn() uint8 {
+	if m.TOS == metaUnavailable {
+		return metaUnavailable
+	}
+	return m.TOS & 0x3
+}
+
+// oobCap fits the two cmsgs (TTL + TOS, each ≤ CMSG_SPACE(4)) with headroom for whatever else
+// the kernel decides to attach.
+const oobCap = 64
+
 // Serve runs the read loop for one socket; call once per bound address.
 // The socket is retained so actions (delayed echo) can pick a family-matching
 // sender later.
@@ -69,14 +109,16 @@ func (s *Server) Serve(conn *net.UDPConn) error {
 	}
 	s.conns = append(s.conns, conn)
 	s.mu.Unlock()
+	enableRecvMeta(conn)
 	buf := make([]byte, 65535)
+	oob := make([]byte, oobCap)
 	for {
-		n, raddr, err := conn.ReadFromUDPAddrPort(buf)
+		n, oobn, _, raddr, err := conn.ReadMsgUDPAddrPort(buf, oob)
 		if err != nil {
 			return err
 		}
 		tRx := time.Since(s.start).Nanoseconds()
-		s.handle(conn, raddr, buf[:n], tRx)
+		s.handle(conn, raddr, buf[:n], tRx, parseMeta(oob[:oobn]))
 	}
 }
 
@@ -126,7 +168,7 @@ func (s *Server) SendDelayedEcho(sess *session.Session, actionID string) error {
 
 // handle enforces spec §3.1/§3.4: unknown prefix, bad HMAC, expired session,
 // replayed seq → silent drop, never a response.
-func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRxNs int64) {
+func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRxNs int64, meta pktMeta) {
 	if len(pkt) < HeaderSize || string(pkt[0:4]) != Magic {
 		return
 	}
@@ -149,6 +191,12 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx
 	if !hmac.Equal(mac.Sum(nil)[:4], pkt[28:32]) {
 		return
 	}
+	// Spec §2.5: over-ceiling traffic is silently dropped (probes tolerate loss by design).
+	// After the HMAC gate so a spoofed flood cannot drain a victim's budget; before the replay
+	// window so a dropped packet's seq stays usable for a resend.
+	if !s.allowUDP(raddr, sess.Device, len(pkt)) {
+		return
+	}
 	if !sess.CheckSeq(seq) {
 		return
 	}
@@ -169,9 +217,16 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx
 		Src: raddr.String(), Size: len(pkt), Type: typ,
 	})
 
+	payload := pkt[HeaderSize : HeaderSize+int(payloadLen)]
 	switch typ {
 	case TypeEchoReq:
-		s.echoResp(conn, raddr, sess, pkt, seq, tRxNs)
+		s.echoResp(conn, raddr, sess, pkt, seq, tRxNs, meta)
+	case TypeTrainData:
+		// No response (spec §3.2): the train is upstream-only; its received view is fetched
+		// afterwards via TRAIN_REPORT_REQ or the observations API.
+		recordTrain(sess, payload, seq, len(pkt), tRxNs, meta)
+	case TypeTrainReportReq:
+		s.trainReport(conn, raddr, sess, payload)
 	case TypeTimesyncReq:
 		s.timesyncResp(conn, raddr, sess, pkt, seq, tRxNs)
 	case TypeMtuProbe:
@@ -181,6 +236,16 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx
 	}
 }
 
+// allowUDP charges the §2.5 packet and byte buckets, per source address and per credential.
+func (s *Server) allowUDP(raddr netip.AddrPort, device string, size int) bool {
+	ipKey, credKey := "ip:"+raddr.Addr().String(), "cred:"+device
+	okA, _ := s.PacketRate.Allow(ipKey)
+	okC, _ := s.PacketRate.Allow(credKey)
+	okAB, _ := s.ByteRate.AllowN(ipKey, float64(size))
+	okCB, _ := s.ByteRate.AllowN(credKey, float64(size))
+	return okA && okC && okAB && okCB
+}
+
 // mtuAck replies to an MTU_PROBE with a small MTU_ACK carrying the total
 // datagram size the server actually received (spec §3.2). The client sends
 // DF-flagged probes of increasing size and binary-searches the path MTU / a
@@ -198,28 +263,43 @@ func (s *Server) mtuAck(conn *net.UDPConn, raddr netip.AddrPort, sess *session.S
 //	 8  8  t_tx_ns
 //	16 16  observed source IP (v4-mapped when v4)
 //	32  2  observed source port
-//	34  1  received TTL   (0xFF = not observed yet; needs recvmsg cmsgs)
+//	34  1  received TTL   (0xFF = not observed; cmsgs unavailable on this host)
 //	35  1  received DSCP/ECN byte (0xFF = not observed)
 //	36  4  received size
-func observation(tRxNs, tTxNs int64, src netip.AddrPort, rcvd int) []byte {
+func observation(tRxNs, tTxNs int64, src netip.AddrPort, rcvd int, meta pktMeta) []byte {
 	b := make([]byte, 40)
 	binary.BigEndian.PutUint64(b[0:8], uint64(tRxNs))
 	binary.BigEndian.PutUint64(b[8:16], uint64(tTxNs))
 	a16 := src.Addr().As16()
 	copy(b[16:32], a16[:])
 	binary.BigEndian.PutUint16(b[32:34], src.Port())
-	b[34], b[35] = 0xFF, 0xFF
+	b[34], b[35] = meta.TTL, meta.TOS
 	binary.BigEndian.PutUint32(b[36:40], uint32(rcvd))
 	return b
 }
 
+// putActionID writes a grant's action id into payload[8:16] — the correlation the spec promises
+// (§5: "an action_id echoed in resulting data-plane packets"), consumed by the client as
+// test.params.action_id (§9). Bytes [0:8] stay with the packet type; [8:16] is reserved for this
+// across every granted type, so the client needs one rule, not five.
+func putActionID(payload []byte, actionID string) {
+	if len(payload) < 16 {
+		return
+	}
+	raw, err := hex.DecodeString(actionID)
+	if err != nil || len(raw) != 8 {
+		return // a malformed id yields zero bytes, not a crash mid-burst
+	}
+	copy(payload[8:16], raw)
+}
+
 // echoResp mirrors the request header (type flipped), appends the observation
 // block, and re-HMACs with the session key. Anti-amplification: the response
 // is capped at the request size (spec §3.4) — the observation block replaces
 // padding rather than growing the datagram; if the request was smaller than
 // header+observation, the block is truncated to fit.
-func (s *Server) echoResp(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, req []byte, seq uint32, tRxNs int64) {
-	obs := observation(tRxNs, time.Since(s.start).Nanoseconds(), raddr, len(req))
+func (s *Server) echoResp(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, req []byte, seq uint32, tRxNs int64, meta pktMeta) {
+	obs := observation(tRxNs, time.Since(s.start).Nanoseconds(), raddr, len(req), meta)
 	max := len(req)
 	if max < HeaderSize {
 		return