fix: compute bitrates over report times, report stats handler errors

docs/guide/client-api.md

@@ -655,7 +655,8 @@ peer.onStats(Duration.ofSeconds(1), stats -> System.out.printf(
 | `candidateType()` | The type of the local candidate in use: `host`, `srflx`, `prflx` or `relay`. Null while not connected. |
 | `roundTripTime()` | The current round trip time, or null while not measured. |
 | `bytesSent()`, `bytesReceived()` | The bytes sent and received on the connection so far. |
-| `outgoingBitrate()`, `incomingBitrate()` | Bits per second since the previous sample, or -1 for a peer's first sample. |
+| `timestamp()` | When WebRTC measured the statistics. |
+| `outgoingBitrate()`, `incomingBitrate()` | Bits per second between the measurement times of the previous sample and this one, or -1 when there is none to compare with: for a peer's first sample, after the clock went back, or after the byte counter started again. A report WebRTC hands out again within its short cache lifetime repeats the previous bitrates. |
 | `packetLoss()` | The share of received packets lost so far, 0 to 1. |
 | `outboundVideo()`, `inboundVideo()` | Codec, resolution and frame rate of each video stream, and what limits a sent one. |
 | `report()` | The whole `RTCStatsReport`. |

webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java

@@ -481,7 +481,7 @@ public Subscription onStats(Duration interval, Consumer<PeerStats> handler) {
 					return;
 				}
 				if (error == null) {
-					handler.accept(stats);
+					call(handler, stats);
 				}
 				else if (!(unwrap(error) instanceof PeerClosedException)) {
 					deliverError(asWebRtcException(error, "Reading the statistics failed"));

webrtc/src/main/java/dev/onvoid/webrtc/client/PeerStats.java

@@ -24,8 +24,8 @@
 
 /**
  * The statistics people look for first, taken from an {@link RTCStatsReport}.
- * Bitrates are computed between two samples, so the first sample a peer takes
- * has none.
+ * Bitrates are computed between the measurement times of two samples, so the
+ * first sample a peer takes has none.
  */
 public final class PeerStats {
 
@@ -71,7 +71,8 @@ public final class PeerStats {
 	}
 
 	/**
-	 * @return when the statistics were taken.
+	 * @return when WebRTC measured the statistics, the timestamp of the
+	 * report.
 	 */
 	public Instant timestamp() {
 		return timestamp;
@@ -107,16 +108,18 @@ public long bytesReceived() {
 	}
 
 	/**
-	 * @return the bits sent per second since the previous sample, or -1 for
-	 * the first sample.
+	 * @return the bits sent per second since the previous sample, or -1 when
+	 * there is none to compare with: for the first sample, after the clock
+	 * went back, or after the byte counter started again.
 	 */
 	public long outgoingBitrate() {
 		return outgoingBitrate;
 	}
 
 	/**
 	 * @return the bits received per second since the previous sample, or -1
-	 * for the first sample.
+	 * when there is none to compare with: for the first sample, after the
+	 * clock went back, or after the byte counter started again.
 	 */
 	public long incomingBitrate() {
 		return incomingBitrate;

webrtc/src/main/java/dev/onvoid/webrtc/client/StatsReader.java

@@ -18,6 +18,7 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -29,16 +30,22 @@
 
 /**
  * Takes {@link PeerStats} from the reports of one peer. It keeps the previous
- * sample to compute bitrates. Used on the peer's control executor.
+ * sample to compute bitrates over the time between the two measurements, not
+ * between their arrival here, which a busy control executor may delay. Used on
+ * the peer's control executor.
  */
 final class StatsReader {
 
-	private long lastNanos = -1;
+	private long lastMicros = -1;
 
 	private long lastBytesSent;
 
 	private long lastBytesReceived;
 
+	private long lastOutgoingBitrate = -1;
+
+	private long lastIncomingBitrate = -1;
+
 
 	PeerStats read(RTCStatsReport report) {
 		Map<String, RTCStats> stats = report.getStats();
@@ -93,29 +100,49 @@ else if (entry.getType() == RTCStatsType.OUTBOUND_RTP) {
 			}
 		}
 
-		long now = System.nanoTime();
+		long micros = report.getTimestamp();
 		long outgoingBitrate = -1;
 		long incomingBitrate = -1;
 
-		if (lastNanos >= 0 && now > lastNanos) {
-			double seconds = (now - lastNanos) / 1e9;
-
-			outgoingBitrate = (long) (Math.max(0, bytesSent - lastBytesSent) * 8 / seconds);
-			incomingBitrate = (long) (Math.max(0, bytesReceived - lastBytesReceived) * 8 / seconds);
+		if (lastMicros >= 0 && micros == lastMicros) {
+			// WebRTC hands out a report again for a short while: the same
+			// measurement, with the same bitrates, measured from the same
+			// sample.
+			outgoingBitrate = lastOutgoingBitrate;
+			incomingBitrate = lastIncomingBitrate;
 		}
+		else {
+			// A clock that went back leaves no interval to measure, and the
+			// next sample is measured from this one.
+			if (lastMicros >= 0 && micros > lastMicros) {
+				double seconds = (micros - lastMicros) / 1e6;
+
+				outgoingBitrate = bitrate(bytesSent, lastBytesSent, seconds);
+				incomingBitrate = bitrate(bytesReceived, lastBytesReceived, seconds);
+			}
 
-		lastNanos = now;
-		lastBytesSent = bytesSent;
-		lastBytesReceived = bytesReceived;
+			lastMicros = micros;
+			lastBytesSent = bytesSent;
+			lastBytesReceived = bytesReceived;
+			lastOutgoingBitrate = outgoingBitrate;
+			lastIncomingBitrate = incomingBitrate;
+		}
 
 		long packetsExpected = packetsReceived + packetsLost;
 		double packetLoss = packetsExpected == 0 ? 0 : (double) packetsLost / packetsExpected;
 
-		return new PeerStats(Instant.now(), candidateType, roundTripTime, bytesSent, bytesReceived,
+		return new PeerStats(Instant.EPOCH.plus(micros, ChronoUnit.MICROS), candidateType, roundTripTime, bytesSent, bytesReceived,
 				outgoingBitrate, incomingBitrate, packetLoss,
 				Collections.unmodifiableList(outbound), Collections.unmodifiableList(inbound), report);
 	}
 
+	private static long bitrate(long bytes, long lastBytes, double seconds) {
+		// Fewer bytes than before: the counter started again, such as on a
+		// transport that replaced another, and what was sent meanwhile is
+		// unknown.
+		return bytes < lastBytes ? -1 : (long) ((bytes - lastBytes) * 8 / seconds);
+	}
+
 	private static VideoStats video(Map<String, RTCStats> stats, Map<String, Object> values,
 			String trackId, String qualityLimitation) {
 		RTCStats codec = lookup(stats, values.get("codecId"));

webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java

@@ -24,6 +24,8 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -551,8 +553,54 @@ void statsAfterConnecting() throws Exception {
 
 		assertEquals("host", first.candidateType());
 		assertTrue(first.bytesSent() > 10_000, () -> "Bytes sent: " + first.bytesSent());
+		assertEquals(first.report().getTimestamp(), ChronoUnit.MICROS.between(Instant.EPOCH, first.timestamp()));
 		assertEquals(-1, first.outgoingBitrate());
-		assertTrue(second.outgoingBitrate() >= 0);
+		// Asked for right after the first, the second may be the report WebRTC
+		// still caches, which repeats the first one's bitrate.
+		if (second.timestamp().equals(first.timestamp())) {
+			assertEquals(-1, second.outgoingBitrate());
+		}
+		else {
+			assertTrue(second.outgoingBitrate() >= 0);
+		}
+	}
+
+	@Test
+	void statsHandlerErrorReported() throws Exception {
+		// A stats handler that throws reaches the runtime's error handler, and
+		// the samples after it still come.
+		ManualTimer timer = new ManualTimer();
+		CompletableFuture<Throwable> reported = new CompletableFuture<>();
+		Duration interval = Duration.ofSeconds(7);
+
+		try (WebRtc runtime = WebRtc.builder()
+				.audio(AudioMode.HEADLESS)
+				.errorHandler(error -> {
+					errors.add(error);
+					reported.complete(error);
+				})
+				.timer(timer)
+				.build()) {
+			Peer peer = runtime.createPeer(manualOptions());
+			AtomicInteger calls = new AtomicInteger();
+			CompletableFuture<PeerStats> later = new CompletableFuture<>();
+
+			peer.onStats(interval, stats -> {
+				if (calls.incrementAndGet() == 1) {
+					throw new IllegalStateException("Handler failed");
+				}
+				later.complete(stats);
+			});
+
+			assertEquals(1, timer.fire(interval));
+			Throwable error = await(reported);
+
+			assertInstanceOf(IllegalStateException.class, error);
+			errors.remove(error);
+
+			assertEquals(1, timer.fire(interval));
+			await(later);
+		}
 	}
 
 	@Test

webrtc/src/test/java/dev/onvoid/webrtc/client/StatsReaderTests.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2026 Alex Andres
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.onvoid.webrtc.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+import dev.onvoid.webrtc.RTCStats;
+import dev.onvoid.webrtc.RTCStatsReport;
+import dev.onvoid.webrtc.RTCStatsType;
+
+import org.junit.jupiter.api.Test;
+
+class StatsReaderTests {
+
+	private static final long START = 1_700_000_000_000_000L;
+
+	private static final long SECOND = 1_000_000;
+
+	private final StatsReader reader = new StatsReader();
+
+
+	@Test
+	void bitrateFromReportTimes() {
+		// Read one right after the other, as a busy control executor hands
+		// them on: the reports were measured a second apart.
+		PeerStats first = reader.read(report(START, 5_000, 2_000));
+		PeerStats second = reader.read(report(START + SECOND, 6_000, 2_500));
+
+		assertEquals(-1, first.outgoingBitrate());
+		assertEquals(-1, first.incomingBitrate());
+		assertEquals(8_000, second.outgoingBitrate());
+		assertEquals(4_000, second.incomingBitrate());
+	}
+
+	@Test
+	void timestampOfReport() {
+		PeerStats stats = reader.read(report(START + 250, 0, 0));
+
+		assertEquals(Instant.ofEpochSecond(1_700_000_000, 250_000), stats.timestamp());
+	}
+
+	@Test
+	void repeatedReportKeepsBitrates() {
+		// WebRTC hands out the report it cached again. The one after it is
+		// measured from the sample before.
+		reader.read(report(START, 0, 0));
+		reader.read(report(START + SECOND, 1_000, 1_000));
+		PeerStats repeated = reader.read(report(START + SECOND, 1_000, 1_000));
+		PeerStats next = reader.read(report(START + 2 * SECOND, 3_000, 1_000));
+
+		assertEquals(8_000, repeated.outgoingBitrate());
+		assertEquals(8_000, repeated.incomingBitrate());
+		assertEquals(16_000, next.outgoingBitrate());
+		assertEquals(0, next.incomingBitrate());
+	}
+
+	@Test
+	void repeatedFirstReport() {
+		reader.read(report(START, 1_000, 1_000));
+		PeerStats repeated = reader.read(report(START, 1_000, 1_000));
+
+		assertEquals(-1, repeated.outgoingBitrate());
+		assertEquals(-1, repeated.incomingBitrate());
+	}
+
+	@Test
+	void clockWentBack() {
+		reader.read(report(START, 0, 0));
+		PeerStats back = reader.read(report(START - SECOND, 1_000, 1_000));
+		PeerStats next = reader.read(report(START, 2_000, 3_000));
+
+		assertEquals(-1, back.outgoingBitrate());
+		assertEquals(-1, back.incomingBitrate());
+		assertEquals(8_000, next.outgoingBitrate());
+		assertEquals(16_000, next.incomingBitrate());
+	}
+
+	@Test
+	void counterStartedAgain() {
+		// Fewer bytes sent than before. Received counts on.
+		reader.read(report(START, 10_000, 1_000));
+		PeerStats reset = reader.read(report(START + SECOND, 500, 2_000));
+		PeerStats next = reader.read(report(START + 2 * SECOND, 1_500, 2_000));
+
+		assertEquals(-1, reset.outgoingBitrate());
+		assertEquals(8_000, reset.incomingBitrate());
+		assertEquals(8_000, next.outgoingBitrate());
+		assertEquals(0, next.incomingBitrate());
+	}
+
+	private static RTCStatsReport report(long micros, long bytesSent, long bytesReceived) {
+		Map<String, Object> values = new HashMap<>();
+		values.put("bytesSent", bytesSent);
+		values.put("bytesReceived", bytesReceived);
+
+		Map<String, RTCStats> stats = new HashMap<>();
+		stats.put("T01", new RTCStats(micros, RTCStatsType.TRANSPORT, "T01", values) { });
+
+		return new RTCStatsReport(stats, micros) { };
+	}
+
+}