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).
func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error) {
	target := sess.DataSource()
	if !target.IsValid() {
		return 0, fmt.Errorf("no observed data-plane source")
	}
	conn := s.connFor(target)
	if conn == nil {
		return 0, fmt.Errorf("no data-plane socket matches target family")
	}
	if sizeBytes < HeaderSize+8 {
		sizeBytes = HeaderSize + 8
	}
	payload := make([]byte, sizeBytes-HeaderSize)
	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)
		}
	}
	return sent, nil
}

// 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)
	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+8 {
				size = HeaderSize + 8
			}
			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))
			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()
}