Take the rate epoch from the slowest stream's frontier, since reading a bucket claims every stream has delivered through it

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"math"
 	"net"
 	"os"
 	"os/signal"
@@ -135,17 +136,20 @@ func (w *rateWindow) at(i int) counterSet {
 	return w.buf[i%len(w.buf)]
 }
 
-// How long the newest epoch may sit still before the wire is taken to have gone
-// quiet. A stamp only advances when a frame arrives, so a frozen epoch means no
-// traffic rather than an unchanged rate.
+// How long the frontier may sit still before the wire is taken to have gone
+// quiet. It only advances when frames arrive on every stream, so a frozen
+// frontier means a stream has stopped delivering rather than an unchanged rate.
 const rateStale = 100 * time.Millisecond
 
-// The newest epoch is still filling, since frames received in it may not have
-// been drained yet, so the rate is read from the one before it.
+// A stamp is only knowable once a worker drains the frame carrying it, so each
+// stream's newest epoch is a frontier: everything the wire delivered to that
+// queue before it has been counted. Reading one bucket across the board takes
+// the one behind the lowest frontier, which every stream has delivered past.
+// The leader's frontier would claim buckets the stragglers are still filling.
 func (d *direction) readRateBucket(now time.Time) {
-	var newest int64
+	newest := int64(math.MaxInt64)
 	for _, r := range d.rxStats {
-		if e := r.newest.Load(); e > newest {
+		if e := r.newest.Load(); e < newest {
 			newest = e
 		}
 	}

main_test.go

@@ -43,7 +43,9 @@ func TestRxBucketWraps(t *testing.T) {
 	}
 }
 
-// The newest epoch may still be filling, so the rate comes from the one before.
+// Every stream's frontier is past the epoch being read, so the bucket is
+// complete across the board and the rate comes from the one before the lowest
+// frontier.
 func TestDirectionRateReadsOneBucketBack(t *testing.T) {
 	d := &direction{rxStats: []*rxStats{{}, {}}}
 	r0 := rateRun{stats: d.rxStats[0]}
@@ -51,6 +53,7 @@ func TestDirectionRateReadsOneBucketBack(t *testing.T) {
 	r0.add(rateBucketNs+1, 500)
 	r1.add(rateBucketNs+2, 700)
 	r0.add(2*rateBucketNs+1, 900)
+	r1.add(2*rateBucketNs+2, 1100)
 	r0.flush()
 	r1.flush()
 
@@ -61,6 +64,46 @@ func TestDirectionRateReadsOneBucketBack(t *testing.T) {
 	}
 }
 
+// Reading a bucket is a claim that every stream has delivered through it, so a
+// stream still short of the epoch holds the read back rather than having its
+// half-filled bucket summed.
+func TestDirectionRateWaitsForSlowestStream(t *testing.T) {
+	d := &direction{rxStats: []*rxStats{{}, {}}}
+	r0 := rateRun{stats: d.rxStats[0]}
+	r1 := rateRun{stats: d.rxStats[1]}
+	r0.add(rateBucketNs+1, 500)
+	r0.add(2*rateBucketNs+1, 900)
+	r0.flush()
+
+	// One stream has never delivered at all, so there is no epoch every stream
+	// has reached and nothing to read.
+	d.readRateBucket(time.Now())
+	if d.rateFrames != 0 || d.rateBytes != 0 {
+		t.Errorf("rate = %d frames, %d bytes; want nothing while a stream has no frontier",
+			d.rateFrames, d.rateBytes)
+	}
+
+	// The straggler is still filling the epoch the leader finished, so its
+	// bucket must not be read yet.
+	r1.add(rateBucketNs+2, 700)
+	r1.flush()
+	d.readRateBucket(time.Now())
+	if d.rateFrames != 0 || d.rateBytes != 0 {
+		t.Errorf("rate = %d frames, %d bytes; want nothing while a stream is still filling the epoch",
+			d.rateFrames, d.rateBytes)
+	}
+
+	// Once it moves past, the bucket is complete for both streams and is read
+	// whole.
+	r1.add(2*rateBucketNs+2, 1100)
+	r1.flush()
+	d.readRateBucket(time.Now())
+	if d.rateFrames != 2 || d.rateBytes != 1200 {
+		t.Errorf("rate = %d frames, %d bytes; want the whole completed epoch, 2 and 1200",
+			d.rateFrames, d.rateBytes)
+	}
+}
+
 // A stamp only advances when a frame arrives, so an epoch that stops moving is
 // a quiet wire and must not keep reporting the last bucket.
 func TestDirectionRateGoesStale(t *testing.T) {