From 892e952a8e01be79be012a69f662a0de627c0142 Mon Sep 17 00:00:00 2001 From: mrambossek Date: Sat, 1 Aug 2026 15:54:09 +0200 Subject: [PATCH] throughput: the upstream direction, counted by the only party that can The client generates the traffic and the server counts it. No grant is involved - the client is sending its own packets, so there is nothing to amplify - but it does need the server's tally, because only the far end knows how much arrived. Without that number a sender measures how fast it can transmit, which is usually just the speed of the local NIC and is not the question being asked. A new wire type the server counts and deliberately never answers: a reply would double the traffic and drag the return path into a measurement that is specifically about the outbound one. The tally is a counter, not a list, and short-circuits before the observation log. A five-second run at 20 Mbps is around ten thousand packets; one struct each would turn a measurement into an allocation storm on a shared server, and nothing needs the per-packet detail since the client holds the send-side record. The gap between the two counts is the loss. direction=up on the throughput action sends nothing - it zeroes the counter, so a second run in one session measures itself instead of inheriting the first. Same honesty rule as downstream: measures_network is false when what arrived matches what was offered, because then the path was never the constraint. Co-Authored-By: Claude Fable 5 --- .../echo_lot/engine/ThroughputMeasurement.kt | 129 ++++++++++++++++++ .../app/echo_lot/protocol/ProbeSession.kt | 43 ++++++ .../main/kotlin/app/echo_lot/protocol/Wire.kt | 7 + server/internal/control/control.go | 27 +++- server/internal/dataplane/udp.go | 12 ++ server/internal/session/session.go | 61 +++++++++ 6 files changed, 277 insertions(+), 2 deletions(-) diff --git a/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt b/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt index 3139cc3..efb4c06 100644 --- a/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt +++ b/echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ThroughputMeasurement.kt @@ -147,6 +147,117 @@ class ThroughputMeasurement(private val ids: IdSource) { ) to findings } + /** + * Upstream throughput: the client sends, the server counts. + * + * The mirror image of the downstream case, and it needs no grant — the client is generating + * its own traffic, so there is no amplification to gate. What it does need is the server's + * count: only the far end knows how much arrived, and without that number a sender can + * measure how fast it can *transmit*, which is not the same question and is usually just the + * speed of the local NIC. + */ + fun runUpstream( + credential: String, + sessionId: String, + control: ControlClient, + probe: ProbeSession, + sessionRef: String, + durationS: Int = 5, + kbps: Int = 20_000, + sizeBytes: Int = 1200, + ): Pair> { + val testId = ids.uuid() + val started = ids.monoNs() + + // Zeroes the server's counter so this run measures itself rather than inheriting the + // packets of an earlier one on the same session. + val reply = runCatching { + control.action(credential, sessionId, """{"action":"throughput","direction":"up"}""") + } + if (reply.isFailure) { + return Test( + id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP, + startedMonoNs = started, endedMonoNs = ids.monoNs(), + status = TestStatus.UNSUPPORTED, + error = TestError("action_refused", reply.exceptionOrNull()?.message ?: "refused"), + ) to emptyList() + } + + val sent = probe.sendThroughput(durationS * 1000L, kbps, sizeBytes) + // A moment for the tail of the run to arrive; counting still-in-flight packets as lost + // would inflate the loss figure by whatever the path's delay happens to be. + Thread.sleep(500) + val seen = upstreamCount(control, credential, sessionId) + + val lossPct = if (sent.packets > 0 && seen != null) { + round2((sent.packets - seen.packets).coerceAtLeast(0) * 100.0 / sent.packets) + } else { + null + } + // The receiver's rate is the measurement. The sender's is what we managed to emit, which + // is a property of this phone and its radio, not of the network. + val achievedKbps = seen?.kbps ?: 0 + + val metrics = json.encodeToJsonElement( + UpstreamThroughputMetrics( + requestedKbps = kbps, + sentPackets = sent.packets, + sentBytes = sent.bytes, + sentKbps = sent.kbps, + receivedPackets = seen?.packets, + receivedBytes = seen?.bytes, + receivedKbps = achievedKbps, + lossPct = lossPct, + // Same honesty rule as downstream: if what arrived matches what we offered, the + // path was never the constraint and this number says nothing about it. + measuresNetwork = seen != null && achievedKbps > 0 && achievedKbps < sent.kbps * 9 / 10, + ), + ) as JsonObject + + val findings = ArrayList() + if (seen != null && seen.packets == 0 && sent.packets > 0) { + findings.add( + finding( + FindingRegistry.THROUGHPUT_NO_DELIVERY, testId, + "No upstream traffic reached the server", + "This device sent ${sent.packets} packets and the server received none. " + + "That is a connectivity fault on the outbound path rather than a slow link.", + ), + ) + } else if (lossPct != null && lossPct >= 2.0) { + findings.add( + finding( + FindingRegistry.THROUGHPUT_BELOW_OFFERED, testId, + "Upstream loss of $lossPct % at ${sent.kbps / 1000} Mbit/s", + "The server received ${seen?.packets} of the ${sent.packets} packets this " + + "device sent. The outbound path could not carry what was offered.", + ), + ) + } + + return Test( + id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP, + startedMonoNs = started, endedMonoNs = ids.monoNs(), + status = if (seen == null || seen.packets == 0) TestStatus.FAILED else TestStatus.OK, + metrics = metrics, + ) to findings + } + + private data class UpstreamCount(val packets: Int, val bytes: Long, val kbps: Int) + + /** The server's tally for this session's upstream run. */ + private fun upstreamCount( + control: ControlClient, credential: String, sessionId: String, + ): UpstreamCount? = runCatching { + val o = Json.parseToJsonElement(control.observations(credential, sessionId)) + .jsonObject["throughput_up"]?.jsonObject ?: return null + UpstreamCount( + packets = o["packets"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0, + bytes = o["bytes"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0, + kbps = o["kbps"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0, + ) + }.getOrNull() + private data class SenderReport( val packets: Int, val bytes: Long, val kbps: Int, val limitedBy: String, ) @@ -186,9 +297,27 @@ class ThroughputMeasurement(private val ids: IdSource) { private fun round2(v: Double) = Math.round(v * 100.0) / 100.0 } +/** Metrics for perf.throughput_udp in the upstream direction. */ +@Serializable +data class UpstreamThroughputMetrics( + val direction: String = "up", + @SerialName("requested_kbps") val requestedKbps: Int, + @SerialName("sent_packets") val sentPackets: Int, + @SerialName("sent_bytes") val sentBytes: Long, + /** What this device managed to emit — a property of the phone and its radio, not the path. */ + @SerialName("sent_kbps") val sentKbps: Int, + @SerialName("received_packets") val receivedPackets: Int? = null, + @SerialName("received_bytes") val receivedBytes: Long? = null, + /** What arrived, measured by the only party that can measure it. This is the result. */ + @SerialName("received_kbps") val receivedKbps: Int, + @SerialName("loss_pct") val lossPct: Double? = null, + @SerialName("measures_network") val measuresNetwork: Boolean, +) + /** Metrics for perf.throughput_udp. */ @Serializable data class ThroughputMetrics( + val direction: String = "down", @SerialName("requested_kbps") val requestedKbps: Int, @SerialName("planned_duration_ms") val plannedDurationMs: Int, @SerialName("packets_received") val packetsReceived: Int, diff --git a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/ProbeSession.kt b/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/ProbeSession.kt index 1ccfc34..997aba8 100644 --- a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/ProbeSession.kt +++ b/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/ProbeSession.kt @@ -109,6 +109,49 @@ class ProbeSession( return out } + /** + * Sends paced upstream traffic for [durationMs] and reports what was put on the wire. + * + * Paced rather than flat out, for the same reason the server paces: an unpaced burst measures + * the local NIC and the first queue it meets, then collapses into loss that reads as a network + * fault. The schedule is absolute rather than sleep-per-packet, which accumulates the + * scheduler's error and drifts the achieved rate below target over a multi-second run. + * + * Nothing comes back — the server counts and stays silent — so the result here is only the + * send side. The measurement is the gap between this and the server's tally. + */ + fun sendThroughput(durationMs: Long, kbps: Int, sizeBytes: Int = 1200): Sent { + val size = sizeBytes.coerceIn(Wire.HEADER_SIZE + 16, 1472) + val payload = ByteArray(size - Wire.HEADER_SIZE) + val perPacketNs = (size.toLong() * 8 * 1_000_000 / kbps.coerceAtLeast(1)).coerceAtLeast(1_000) + + val start = System.nanoTime() + val deadline = start + durationMs * 1_000_000 + var next = start + var packets = 0 + var bytes = 0L + while (System.nanoTime() < deadline) { + val pkt = Wire.build(Wire.TYPE_THROUGHPUT_UP, prefix, ++seq, nowNs(), key, payload) + try { + socket.send(DatagramPacket(pkt, pkt.size, server)) + } catch (e: java.io.IOException) { + // A local send failure is our condition, not the path's. Stop and report what + // actually left, rather than counting the remainder as loss on the network. + break + } + packets++ + bytes += pkt.size + next += perPacketNs + val sleepNs = next - System.nanoTime() + if (sleepNs > 0) Thread.sleep(sleepNs / 1_000_000, (sleepNs % 1_000_000).toInt()) + } + val elapsedMs = (System.nanoTime() - start) / 1_000_000 + return Sent(packets, bytes, elapsedMs, if (elapsedMs > 0) (bytes * 8 / elapsedMs).toInt() else 0) + } + + /** What one upstream run put on the wire locally. */ + data class Sent(val packets: Int, val bytes: Long, val durationMs: Long, val kbps: Int) + /** One packet received from the server, with the wire size actually delivered. */ data class Received(val type: Int, val seq: Int, val sizeBytes: Int, val tRxNs: Long) diff --git a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt b/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt index 5f9c46c..93c3567 100644 --- a/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt +++ b/echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt @@ -41,6 +41,13 @@ object Wire { /** One packet of a sustained-rate downstream run. */ const val TYPE_THROUGHPUT_DATA: Int = 0x0E + /** + * One packet of a client-driven upstream run. The server counts it and does not answer: + * a reply would double the traffic and drag the return path into a measurement that is + * specifically about the outbound one. + */ + const val TYPE_THROUGHPUT_UP: Int = 0x0F + /** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */ fun wirePrefix(sessionId: String): ByteArray { require(sessionId.length >= 16) { "session id too short" } diff --git a/server/internal/control/control.go b/server/internal/control/control.go index 9e95186..77a8cb4 100644 --- a/server/internal/control/control.go +++ b/server/internal/control/control.go @@ -225,7 +225,9 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) { "connect_back": cb, // The sender's own count, which is what makes the receiver's count mean something. "throughput": sess.ThroughputReports(), - "dns_canary": dnsCanary, + // The receiver's count for upstream runs — same idea, other direction. + "throughput_up": upstreamJSON(sess), + "dns_canary": dnsCanary, // TODO(spec §6): http echo records }) } @@ -433,9 +435,20 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { // Only the downstream direction needs the server to send. Upstream is the client // sending and the server counting, which needs no action at all — so asking for it here // is a client bug worth naming rather than silently doing the other thing. + if req.Direction == "up" { + // Upstream needs nothing sent from here — the client generates the traffic and the + // server counts it. The only thing an action can usefully do is zero the counter so + // the run measures itself rather than inheriting an earlier one. + sess.ResetUpstream() + writeJSON(w, http.StatusAccepted, map[string]any{ + "action_id": actionID, "direction": "up", "reset": true, + "note": "send TYPE_THROUGHPUT_UP packets, then read observations.throughput_up", + }) + return + } if req.Direction != "" && req.Direction != "down" { writeJSON(w, http.StatusBadRequest, map[string]string{ - "error": "only direction=down is an action; upstream throughput is measured by sending and reading observations", + "error": "direction must be up or down", }) return } @@ -785,3 +798,13 @@ func (s *Server) EnrollmentLink(token string) string { "&p=" + url.QueryEscape("pin-sha256:"+s.PinB64) + "&t=" + url.QueryEscape(token) } + +// upstreamJSON renders the upstream tally with the derived figures already computed, so every +// consumer does not have to repeat (and risk fumbling) the same arithmetic. +func upstreamJSON(sess *session.Session) map[string]any { + u := sess.Upstream() + return map[string]any{ + "packets": u.Packets, "bytes": u.Bytes, + "span_ms": u.SpanMs(), "kbps": u.Kbps(), + } +} diff --git a/server/internal/dataplane/udp.go b/server/internal/dataplane/udp.go index 879ff0e..e571796 100644 --- a/server/internal/dataplane/udp.go +++ b/server/internal/dataplane/udp.go @@ -39,6 +39,10 @@ const ( TypeFragData = 0x0D // TypeThroughputData is one packet of a sustained-rate downstream run. TypeThroughputData = 0x0E + // TypeThroughputUp is one packet of a client-driven upstream run. The server counts it and + // deliberately does not answer: a reply would double the traffic and measure the return + // path at the same time, which is the one thing this test is trying not to do. + TypeThroughputUp = 0x0F ) type Server struct { @@ -152,6 +156,14 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx if la, ok := conn.LocalAddr().(*net.UDPAddr); ok { sess.NoteDataLocal(la.AddrPort()) } + + // Upstream throughput short-circuits before the observation log. Recording one struct per + // packet here would mean tens of thousands of allocations for a single run; the counter is + // all anyone needs, since the client holds the send-side record. + if typ == TypeThroughputUp { + sess.CountUpstream(len(pkt), tRxNs) + return + } sess.RecordUDP(session.UDPObservation{ Seq: seq, TRxNs: tRxNs, TTxNs: time.Since(s.start).Nanoseconds(), Src: raddr.String(), Size: len(pkt), Type: typ, diff --git a/server/internal/session/session.go b/server/internal/session/session.go index cbf8046..b073c86 100644 --- a/server/internal/session/session.go +++ b/server/internal/session/session.go @@ -41,6 +41,7 @@ type Session struct { udpObs []UDPObservation // ring, newest last, cap obsCap connectBack []ConnectBackResult throughput []ThroughputReport + upstream UpstreamCounter } const obsCap = 4096 @@ -67,6 +68,36 @@ type ThroughputReport struct { LimitedBy string `json:"limited_by"` } +// UpstreamCounter is the server's tally of a client-driven throughput run. +// +// Deliberately a counter and not a list. A five-second upstream run at 20 Mbps is around ten +// thousand packets; one observation struct each would turn a measurement into an allocation +// storm on a shared server, and nothing downstream needs the per-packet detail - the client +// already has its own send record. The gap between the two counts IS the loss. +type UpstreamCounter struct { + Packets int `json:"packets"` + Bytes int64 `json:"bytes"` + FirstRxNs int64 `json:"first_rx_ns"` + LastRxNs int64 `json:"last_rx_ns"` +} + +// SpanMs is the time between the first and last packet, which is the interval the rate should be +// computed over - not the client's requested duration, which includes ramp-up and the tail. +func (u UpstreamCounter) SpanMs() int64 { + if u.Packets < 2 || u.LastRxNs <= u.FirstRxNs { + return 0 + } + return (u.LastRxNs - u.FirstRxNs) / 1_000_000 +} + +// Kbps is bits per millisecond, which is kilobits per second - no scaling constant to get wrong. +func (u UpstreamCounter) Kbps() int { + if ms := u.SpanMs(); ms > 0 { + return int(u.Bytes * 8 / ms) + } + return 0 +} + // ConnectBackResult records one connect-back action outcome. type ConnectBackResult struct { ActionID string `json:"action_id"` @@ -100,6 +131,36 @@ func (s *Session) Observations() (packetsSeen uint64, udp []UDPObservation, cb [ append([]ConnectBackResult(nil), s.connectBack...) } +// CountUpstream tallies one client-sent throughput packet. +// +// Called on the hot path for every packet of an upstream run, so it does exactly two additions +// and two comparisons under the lock and allocates nothing. +func (s *Session) CountUpstream(sizeBytes int, tRxNs int64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.upstream.Packets == 0 { + s.upstream.FirstRxNs = tRxNs + } + s.upstream.Packets++ + s.upstream.Bytes += int64(sizeBytes) + s.upstream.LastRxNs = tRxNs +} + +// Upstream returns the tally so far. +func (s *Session) Upstream() UpstreamCounter { + s.mu.Lock() + defer s.mu.Unlock() + return s.upstream +} + +// ResetUpstream clears the tally, so a second run in one session measures itself rather than +// inheriting the first one's packets. +func (s *Session) ResetUpstream() { + s.mu.Lock() + defer s.mu.Unlock() + s.upstream = UpstreamCounter{} +} + // RecordThroughput stores the server's account of one sustained send. // // Kept as a per-action summary rather than per-packet records: a ten-second run at 50 Mbps is