echolot/server/internal/dataplane/granted.go

// SPDX-FileCopyrightText: 2026 Echolot contributors
// SPDX-License-Identifier: GPL-3.0-or-later

package dataplane

import (
	"encoding/binary"
	"fmt"
	"time"

	"echo-lot.app/server/internal/session"
)

// Server-to-client sends that exceed the request size, and are therefore only legal under an
// asymmetric grant (spec §3.4). Every one of them:
//   - targets the session's observed data-plane source, fixed when the grant was created;
//   - stops the moment the grant's byte/rate/time budget is exhausted;
//   - is HMAC-signed with the session key, so the client can tell our packets from injected ones.

// DownTrain sends `count` DOWNTRAIN_DATA packets of `sizeBytes` spaced `intervalUs` apart. The
// 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).
//
// 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")
	}
	conn := s.connFor(target, sess.DataLocal())
	if conn == nil {
		return 0, fmt.Errorf("no data-plane socket matches target family")
	}
	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
	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
	}
	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
// Err means *we* could not put it on the wire (the datagram exceeds our own egress MTU with DF
// set) — the client must not read its absence as a path limit, so this is reported, not hidden.
type BigSendResult struct {
	SizeBytes int    `json:"size_bytes"`
	Seq       int    `json:"seq"`
	Sent      bool   `json:"sent"`
	Err       string `json:"err,omitempty"`
}

// BigSend transmits one datagram per requested size so the client can see which sizes survive the
// *downstream* path — the mtu.pmtud_down / mtu.frag_delivery evidence. The client cannot produce
// this itself: only the far end can emit a large packet toward it.
//
// With df set, the DF bit is forced for the whole burst, so nothing fragments and the largest
// size that arrives IS the downstream path MTU. Without it, the kernel fragments freely and the
// result only says whether fragments get through — a different (also useful) measurement, and
// the reason the two are separate test types.
func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, df bool) ([]BigSendResult, error) {
	target := sess.DataSource()
	if !target.IsValid() {
		return nil, fmt.Errorf("no observed data-plane source")
	}
	conn := s.connFor(target, sess.DataLocal())
	if conn == nil {
		return nil, fmt.Errorf("no data-plane socket matches target family")
	}

	results := make([]BigSendResult, 0, len(sizes))
	burst := func() error {
		for i, size := range sizes {
			if size < HeaderSize+16 {
				size = HeaderSize + 16
			}
			if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway
				size = 9000
			}
			if !g.Allow(size) {
				break
			}
			payload := make([]byte, size-HeaderSize)
			// 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),
			})
			time.Sleep(20 * time.Millisecond) // keep bursts from being read as congestion loss
		}
		return nil
	}

	if df && dfSupported {
		s.dfMu.Lock()
		defer s.dfMu.Unlock()
		if err := withDF(conn, burst); err != nil {
			return results, err
		}
		return results, nil
	}
	return results, burst()
}

func errString(err error) string {
	if err == nil {
		return ""
	}
	return err.Error()
}