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 <noreply@anthropic.com>

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<Test, List<Finding>> {
+        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<Finding>()
+        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,

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)
 

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" }

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(),
+	}
+}

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,

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