ipcpd: Add flow overshoot protection to mb-ecntestingbe

The ECN marks were the only source of congestion information, and
going into congestion collapse (dropping a lot of ECN packets and
causing loss of signal) was unrecoverable.

The sender's slow-start ramp clock now tracks a measured RTT to avoid
overshooting into congestion collapse on the previously fixed (14ms)
doubling on long-delay (e.g. 100ms) links.

The flow allocator now carries a periodic heartbeat/ack to keep RTT
estimates up-to-date. The heartbeat also serves as a congestion
collapse detector: consecutive heartbeat losses will go into loss
recovery, halving the window and pausing Additive Increase until the
path is clear. The DT component now has a max_rtt config parameter that
serves as a hint for the layer RTT to optimize slow-start for
low-latency networks.

These two mechanisms follow TCP, however only during slow start up to
the first ECN signal and to recover from a total collapse until the
ECN signal returns. MB-ECN is still fully RTT-independent in the AI/PD
region. A shorter RTT path will just reach its equal fair share faster
than a longer RTT path.

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

16 files changed, 879 insertions, 83 deletions

diff --git a/include/ouroboros/ipcp.h b/include/ouroboros/ipcp.h
index e29b080a..135b8fcb 100644
--- a/include/ouroboros/ipcp.h
+++ b/include/ouroboros/ipcp.h
@@ -124,9 +124,10 @@ enum pol_cong_avoid {
 
 struct dt_config {
         struct {
-                uint8_t addr_size;
-                uint8_t eid_size;
-                uint8_t max_ttl;
+                uint8_t  addr_size;
+                uint8_t  eid_size;
+                uint8_t  max_ttl;
+                uint16_t max_rtt; /* Declared max layer RTT (ms)       */
         };
         struct routing_config routing; /* Routing policy */
 };
@@ -135,6 +136,7 @@ static const struct dt_config default_dt_config = {
         .addr_size = 4,
         .eid_size  = 8,
         .max_ttl   = 60,
+        .max_rtt   = 200,
         .routing = {
                 .pol = ROUTING_LINK_STATE,
                 .ls = {
diff --git a/irmd.conf.in b/irmd.conf.in
index b9b79782..02cb1da3 100644
--- a/irmd.conf.in
+++ b/irmd.conf.in
@@ -113,6 +113,7 @@ autobind=true           # Defaults to false.
 # addr_size=4           # Defaults to 4 (32-bit addresses).
 # eid_size=8            # Defaults to 8 (64-bit endpoint IDs, only accepted option).
 # max_ttl=60            # Defaults to 60 (max 255).
+# max_rtt=200           # Defaults to 200 (max 65535).
 # addr-auth="flat"      # Defaults to flat (currently only option).
 # routing="lfa"         # Defaults to link-state (options: link-state, lfa, ecmp).
 # ls_t_recalc=4         # Forwarding Function update interval (s) (Default 4).
diff --git a/src/ipcpd/unicast/ca.c b/src/ipcpd/unicast/ca.c
index d0ee2f73..f2a65fff 100644
--- a/src/ipcpd/unicast/ca.c
+++ b/src/ipcpd/unicast/ca.c
@@ -58,7 +58,8 @@ struct {
 #endif
 } ca;
 
-int ca_init(enum pol_cong_avoid pol)
+int ca_init(enum pol_cong_avoid pol,
+            uint32_t            rtt_ms)
 {
 #ifndef IPCP_CA_PER_FLOW
         size_t i;
@@ -71,6 +72,7 @@ int ca_init(enum pol_cong_avoid pol)
         case CA_MB_ECN:
                 log_dbg("Using multi-bit ECN.");
                 ca.ops = &mb_ecn_ca_ops;
+                mb_ecn_init(rtt_ms);
                 break;
         default:
                 return -1;
@@ -216,6 +218,29 @@ void ca_ctx_update_ece(void *   _ctx,
         return ca.ops->ctx_update_ece(ctx->pol, ece, cap);
 }
 
+bool ca_ctx_hb_due(void *   _ctx,
+                   uint64_t now)
+{
+        struct ca_ctx * ctx = _ctx;
+
+        if (ca.ops->ctx_hb_due == NULL)
+                return false;
+
+        return ca.ops->ctx_hb_due(ctx->pol, now);
+}
+
+void ca_ctx_rtt(void *   _ctx,
+                uint64_t now,
+                uint64_t rtt)
+{
+        struct ca_ctx * ctx = _ctx;
+
+        if (ca.ops->ctx_rtt == NULL)
+                return;
+
+        ca.ops->ctx_rtt(ctx->pol, now, rtt);
+}
+
 int  ca_calc_ecn(size_t    queued,
                  uint8_t * ecn,
                  qoscube_t qc,
diff --git a/src/ipcpd/unicast/ca.h b/src/ipcpd/unicast/ca.h
index d73d35f5..fc6de445 100644
--- a/src/ipcpd/unicast/ca.h
+++ b/src/ipcpd/unicast/ca.h
@@ -29,7 +29,11 @@
 #include <stdbool.h>
 #include <sys/types.h>
 
-int      ca_init(enum pol_cong_avoid ca);
+/* Buffer a policy's ca_print_stats output must fit in. */
+#define CA_STATS_STRLEN 2048
+
+int      ca_init(enum pol_cong_avoid ca,
+                 uint32_t            rtt_ms);
 
 void     ca_fini(void);
 
@@ -56,6 +60,13 @@ void     ca_ctx_update_ece(void *   ctx,
                            uint16_t ece,
                            uint8_t  cap);
 
+bool     ca_ctx_hb_due(void *   ctx,
+                       uint64_t now);
+
+void     ca_ctx_rtt(void *   ctx,
+                    uint64_t now,
+                    uint64_t rtt);
+
 int      ca_calc_ecn(size_t    queued,
                      uint8_t * ecn,
                      qoscube_t qc,
diff --git a/src/ipcpd/unicast/ca/mb-ecn.c b/src/ipcpd/unicast/ca/mb-ecn.c
index e59aac88..a4c9f29e 100644
--- a/src/ipcpd/unicast/ca/mb-ecn.c
+++ b/src/ipcpd/unicast/ca/mb-ecn.c
@@ -40,14 +40,24 @@
 #include <stdio.h>
 
 /*
- * The sender paces with a token bucket at a rate driven by Δt-scaled
- * AIMD: each step changes the rate proportional to elapsed wall-clock
- * time, so the per-second dynamics do not depend on how often packets
- * arrive. The receiver's averaging window and the sender's feedback
- * staleness both stretch with the flow's byte rate, so a slow flow
- * is measured and controlled like a fast one; CA_RATE_MIN only
- * bounds those horizons (window <= CA_TW_ABSMAX, TTL ~8 s). There
- * is no per-flow timer; the control runs on packet sends.
+ * Multi-bit ECN congestion avoidance: a rate-based controller. The
+ * sender paces a token bucket at a rate steered by graded ECN
+ * feedback, so the backoff is proportional to the congestion. A
+ * backlogged flow ramps in slow start to find the path capacity, then
+ * settles into AIMD around its fair share. There is no sliding window
+ * and no per-flow timer; the control runs on sends.
+ *
+ * Every rate step is scaled by elapsed wall-clock time (Δt), not by
+ * packet count, so the per-second dynamics are RTT-independent. The
+ * receiver's averaging window and the sender's feedback staleness both
+ * stretch with the flow's byte rate, so a slow flow is measured and
+ * controlled like a fast one; CA_RATE_MIN only bounds those horizons
+ * (window <= CA_TW_ABSMAX, TTL ~8 s).
+ *
+ * The ramp clock ss_tc = 2 * RTT holds slow-start overshoot near 1.65x
+ * (e^1/2): it seeds from the declared max_rtt and then tracks the
+ * heartbeat's measured RTT. Feedback silence past the staleness horizon
+ * leaves slow start; a sustained run of it restarts at the floor.
  *
  * The floor and the AI slope scale with the path: forwarders stamp
  * their measured link capacity into the PCI (cap.c), the receiver
@@ -77,12 +87,17 @@
 #define CA_PROBE_TC  (8ULL * BILLION)     /* proportional probe TC 8s   */
 #define CA_ECE_REF   (16 << CA_SHFT)      /* full congestion: ecn 16    */
 #define CA_MD_KD_DIV 4                    /* one-sided lead gain 1/4    */
-/* Must stay >= 1 ms: the MD term scales by dtc / MILLION (truncates).  */
-#define CA_DT_CTRL   (BILLION / 1000)
+#define CA_DT_CTRL   (BILLION / 1000)     /* min rate-update spacing    */
 #define CA_DT_CAP    (BILLION / 20)       /* idle-resume Δt clamp 50ms  */
 /* Floor of the rate-relative feedback staleness (ctx->ece_ttl). */
 #define CA_ECE_TTL   ((1 << CA_TW_GAP_SHFT) * CA_TW_INIT)
-#define CA_SS_TC     (BILLION / 50)
+#define CA_SS_RTT_MUL 2                   /* ss_tc = 2 * layer RTT      */
+#define CA_SS_TC_MIN (BILLION / 1000)     /* ramp floor 1 ms            */
+#define CA_SS_TC_MAX (4ULL * BILLION)     /* ramp ceiling 4 s           */
+#define CA_HB_MIN    (40 * MILLION)       /* heartbeat interval floor   */
+#define CA_HB_LOSS   4                    /* stale horizons -> restart  */
+#define CA_RTT_SHFT  2                    /* ss_tc EWMA weight 1/4      */
+#define CA_SS_RTT_DEF 200                 /* default layer RTT (ms)     */
 #define CA_WASH_BKT  (BILLION / 32)       /* washout bucket ~31 ms      */
 #define CA_WASH_SHFT 2                    /* damp 1/4 of bucket change  */
 
@@ -126,31 +141,29 @@ struct mb_ecn_ctx {
         uint16_t tx_ece;      /* congestion reported from downstream  */
         uint16_t tx_ecp;      /* previous tx_ece (rise detection)     */
         uint8_t  tx_loc;      /* local first-hop ecn mark (fallback)  */
-        uint8_t  tx_cap;      /* path capacity code fed back to us    */
         bool     tx_cav;      /* past slow start                      */
+        bool     ai_hold;     /* freeze AI after loss until clear     */
         uint64_t rate;        /* paced send rate (bytes/s)            */
         uint64_t rate_min;    /* capacity-derived rate floor (B/s)    */
         uint64_t ai_rate;     /* additive-increase slope (B/s^2)      */
         uint64_t ece_ttl;     /* how long feedback stays valid (ns)   */
+        uint64_t ss_tc;       /* slow-start time constant (ns)        */
         uint64_t r_bkt;       /* rate snapshot at last washout bucket */
         uint64_t wash_acc;    /* washout bucket time accumulator (ns) */
+        uint64_t dec_acc;     /* sub-ms decrease time carried (ns)    */
         uint64_t inv_rate;    /* fixed-point 1/rate for pacing        */
         uint64_t vt;          /* virtual service clock (bytes)        */
         uint64_t lead;        /* pacer lead of last send (bytes)      */
         uint64_t last_ts;     /* last clock advance (ns)              */
         uint64_t last_ctrl;   /* last rate update (ns)                */
-        uint64_t last_fb;     /* last feedback applied (ns)           */
+        uint64_t last_fb;     /* last congestion feedback (ns)        */
+        uint64_t last_sig;    /* last liveness signal, incl. hb (ns)  */
+        uint64_t n_fb;        /* feedback updates received            */
+        uint64_t last_hb;     /* last heartbeat emitted (ns)          */
+        uint64_t last_res;    /* last resume from idle (ns)           */
         uint64_t last_loc;    /* last local mark seen (ns)            */
         uint64_t last_cap;    /* last capacity applied (ns)           */
 
-        uint64_t n_ctrl;      /* control steps taken                  */
-        uint64_t t_ctrl;      /* wall time covered by steps (ns)      */
-        uint64_t t_bank;      /* increase time banked in steps (ns)   */
-        uint64_t n_fb;        /* feedback updates received            */
-        uint64_t n_ttl;       /* feedback aged out (TTL)              */
-        uint64_t n_cap;       /* capacity updates applied             */
-        uint64_t ss_peak;     /* peak rate in slow start (bytes/s)    */
-
         uint64_t snd_byt;     /* bytes offered this window (capped)   */
         uint64_t snd_win;     /* utilisation window start (ns)        */
         uint64_t snd_r0;      /* rate at window start                 */
@@ -158,14 +171,31 @@ struct mb_ecn_ctx {
         bool     backlogged;  /* offered load keeps the pacer busy    */
         bool     src_limited; /* rate held at offered-load ceiling    */
         bool     started;     /* a real send has occurred             */
+
+        /* Diagnostics only, read by mb_ecn_print_stats. */
+        uint8_t  tx_cap;      /* path capacity code fed back to us    */
+        uint64_t n_ctrl;      /* control steps taken                  */
+        uint64_t t_ctrl;      /* wall time covered by steps (ns)      */
+        uint64_t t_bank;      /* increase time banked in steps (ns)   */
+        uint64_t n_ttl;       /* feedback aged out (TTL)              */
+        uint64_t n_cap;       /* capacity updates applied             */
+        uint64_t n_loss;      /* signal-loss cuts (collapse)          */
+        uint64_t n_rtt;       /* heartbeat RTT samples folded         */
+        uint64_t ss_peak;     /* peak rate in slow start (bytes/s)    */
 };
 
+/* Layer slow-start time constant (ns), from the declared RTT. */
+static uint64_t mb_ecn_ss_tc = (uint64_t) CA_SS_RTT_MUL *
+                               CA_SS_RTT_DEF * MILLION;
+
 struct ca_ops mb_ecn_ca_ops = {
         .ctx_create     = mb_ecn_ctx_create,
         .ctx_destroy    = mb_ecn_ctx_destroy,
         .ctx_update_snd = mb_ecn_ctx_update_snd,
         .ctx_update_rcv = mb_ecn_ctx_update_rcv,
         .ctx_update_ece = mb_ecn_ctx_update_ece,
+        .ctx_hb_due     = mb_ecn_ctx_hb_due,
+        .ctx_rtt        = mb_ecn_ctx_rtt,
         .calc_ecn       = mb_ecn_calc_ecn,
         .marks_ecn      = true,
         .print_stats    = mb_ecn_print_stats
@@ -190,6 +220,21 @@ static uint64_t mb_ecn_ece_ttl(uint64_t rate)
         return ttl > (uint64_t) CA_ECE_TTL ? ttl : (uint64_t) CA_ECE_TTL;
 }
 
+/* Derive the layer slow-start slope from the declared RTT (ms). */
+void mb_ecn_init(uint32_t rtt_ms)
+{
+        uint64_t tc;
+
+        if (rtt_ms == 0)          /* unspecified: safe default */
+                rtt_ms = CA_SS_RTT_DEF;
+
+        tc = (uint64_t) CA_SS_RTT_MUL * rtt_ms * MILLION;
+        if (tc < (uint64_t) CA_SS_TC_MIN)
+                tc = CA_SS_TC_MIN;
+
+        mb_ecn_ss_tc = tc;
+}
+
 void * mb_ecn_ctx_create(void)
 {
         struct timespec     now;
@@ -209,6 +254,7 @@ void * mb_ecn_ctx_create(void)
         ctx->rate      = CA_RATE_INIT;
         ctx->rate_min  = CA_RATE_MIN;
         ctx->ai_rate   = CA_AI_RATE;
+        ctx->ss_tc     = mb_ecn_ss_tc;
         ctx->ece_ttl   = mb_ecn_ece_ttl(CA_RATE_INIT);
         ctx->r_bkt     = CA_RATE_INIT;
         ctx->inv_rate  = mb_ecn_rate_inv(CA_RATE_INIT);
@@ -218,6 +264,7 @@ void * mb_ecn_ctx_create(void)
         ctx->last_ts   = t;
         ctx->last_ctrl = t;
         ctx->last_fb   = t;
+        ctx->last_sig  = t;
         ctx->last_loc  = t;
         ctx->last_cap  = t;
 
@@ -252,7 +299,7 @@ static void mb_ecn_slow_start(struct mb_ecn_ctx * ctx,
                               uint64_t            dta)
 {
         if (ctx->backlogged)
-                ctx->rate += ctx->rate * dta / CA_SS_TC;
+                ctx->rate += ctx->rate * dta / ctx->ss_tc;
 
         ctx->r_bkt = ctx->rate;
 }
@@ -264,11 +311,18 @@ static void mb_ecn_increase(struct mb_ecn_ctx * ctx,
         if (!ctx->backlogged)
                 return;
 
+        /* After a loss, hold until a clean signal drains the queue. */
+        if (ctx->ai_hold)
+                return;
+
         ctx->rate += ctx->ai_rate * dta / BILLION;
         ctx->rate += ctx->rate * dta / CA_PROBE_TC;
 }
 
-/* Multiplicative decrease: proportional cut plus a one-sided lead. */
+/*
+ * Multiplicative decrease: cut proportional to mark x elapsed time,
+ * plus a one-sided lead that cuts extra while the mark is rising.
+ */
 static void mb_ecn_decrease(struct mb_ecn_ctx * ctx,
                             uint64_t            dtc)
 {
@@ -281,19 +335,21 @@ static void mb_ecn_decrease(struct mb_ecn_ctx * ctx,
         m = ctx->tx_ece > 0 ? ctx->tx_ece
                             : (uint16_t) (ctx->tx_loc << CA_SHFT);
         if (m == 0) {
-                ctx->tx_ecp = 0;
+                ctx->dec_acc = 0;      /* unmarked time is not banked */
+                ctx->tx_ecp  = 0;
                 return;
         }
 
         mark = MIN(m, CA_ECE_REF);
-        rise = 0;
-        if (m > ctx->tx_ecp)
-                rise = MIN(m - ctx->tx_ecp, CA_ECE_REF);
 
-        cut = ctx->rate * rise / (CA_ECE_REF * CA_MD_KD_DIV);
+        /* One-sided lead: cut extra while the mark is still rising. */
+        rise = m > ctx->tx_ecp ? MIN(m - ctx->tx_ecp, CA_ECE_REF) : 0;
+        cut  = ctx->rate * rise / (CA_ECE_REF * CA_MD_KD_DIV);
 
-        /* Honest elapsed ms, so a starved sender still cuts. */
-        dtm = dtc / MILLION;
+        /* Honest elapsed ms; the sub-ms remainder carries over. */
+        ctx->dec_acc += dtc;
+        dtm = ctx->dec_acc / MILLION;
+        ctx->dec_acc -= dtm * MILLION;
         if (mark * dtm >= CA_ECE_REF * 500)
                 cut += ctx->rate / 2;
         else
@@ -428,11 +484,15 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx,
 {
         uint64_t elapsed = t - ctx->snd_win;
         uint64_t offered;
+        bool     was = ctx->backlogged;
 
         offered = ctx->snd_byt * BILLION / elapsed;
 
         ctx->backlogged = offered * CA_USE_DEN >= ctx->snd_r0 * CA_USE_NUM;
 
+        if (!was && ctx->backlogged)   /* resume: fresh liveness baseline */
+                ctx->last_res = t;
+
         mb_ecn_offered(ctx, offered, elapsed);
 
         if (ctx->backlogged)
@@ -444,9 +504,50 @@ static void mb_ecn_win(struct mb_ecn_ctx * ctx,
 }
 
 /* Age out congestion, local-mark and capacity signals once stale. */
+/* Heartbeat interval: ~1 RTT, floored so fast links don't over-probe. */
+static uint64_t mb_ecn_t_hb(const struct mb_ecn_ctx * ctx)
+{
+        uint64_t t = ctx->ss_tc >> 1;
+
+        return t > (uint64_t) CA_HB_MIN ? t : CA_HB_MIN;
+}
+
+/* Feedback collapsed while backlogged: halve like an RTO, stay in AIMD. */
+static void mb_ecn_loss(struct mb_ecn_ctx * ctx,
+                        uint64_t            t)
+{
+        ctx->rate -= ctx->rate / 2;
+        if (ctx->rate < (uint64_t) CA_RATE_MIN)
+                ctx->rate = CA_RATE_MIN;
+
+        ctx->r_bkt    = ctx->rate;
+        ctx->inv_rate = mb_ecn_rate_inv(ctx->rate);
+        ctx->ece_ttl  = mb_ecn_ece_ttl(ctx->rate);
+        ctx->last_sig = t;
+        ctx->ai_hold  = true;
+        ctx->n_loss++;
+}
+
 static void mb_ecn_age(struct mb_ecn_ctx * ctx,
                        uint64_t            t)
 {
+        uint64_t ttl = ctx->ece_ttl;
+        uint64_t ref = ctx->last_sig > ctx->last_res
+                       ? ctx->last_sig : ctx->last_res;
+        uint64_t gap = t - ref;
+
+        /*
+         * Sustained silence while backlogged is feedback collapse: cut
+         * the rate in half and stay in AIMD, so a recovering flow climbs
+         * back additively instead of re-ramping. Repeated silence decays
+         * it geometrically toward the floor.
+         */
+        if (ctx->backlogged && ctx->n_fb + ctx->n_rtt > 0
+            && gap > (uint64_t) CA_HB_LOSS * ttl) {
+                mb_ecn_loss(ctx, t);
+                return;
+        }
+
         if (t - ctx->last_fb > ctx->ece_ttl) {
                 if (ctx->tx_ece > 0)
                         ctx->n_ttl++;
@@ -505,15 +606,20 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
 
         /* Lazy warm-up seed: packet #1 is never an idle resume. */
         if (!ctx->started) {
-                ctx->started = true;
-                ctx->last_ts = t;
-                ctx->snd_win = t;
-                ctx->snd_r0  = ctx->rate;
+                ctx->started  = true;
+                ctx->last_ts  = t;
+                ctx->last_res = t;
+                ctx->snd_win  = t;
+                ctx->snd_r0   = ctx->rate;
         }
 
         dt = t - ctx->last_ts;
         ctx->last_ts = t;
 
+        /* Idle gap clears backlog before aging: no false loss on resume. */
+        if (dt > (uint64_t) CA_DT_CAP)
+                ctx->backlogged = false;
+
         mb_ecn_age(ctx, t);
 
         /* Offered-load estimator: accumulate, gate growth, size ceiling. */
@@ -521,9 +627,6 @@ static time_t mb_ecn_snd(struct mb_ecn_ctx * ctx,
         if (ctx->snd_byt > (uint64_t) CA_SND_BYT_MAX)
                 ctx->snd_byt = CA_SND_BYT_MAX;
 
-        if (dt > (uint64_t) CA_DT_CAP)
-                ctx->backlogged = false;
-
         if (t - ctx->snd_win >= (uint64_t) CA_SND_WIN)
                 mb_ecn_win(ctx, t);
 
@@ -683,8 +786,14 @@ static void mb_ecn_ece(struct mb_ecn_ctx * ctx,
         uint64_t tgt;
 
         ctx->tx_ece  = ece;
-        ctx->tx_cav  = true;
-        ctx->last_fb = t;
+        ctx->tx_cav  = true;   /* closed-loop feedback: leave slow start */
+
+        /* A clean (unsaturated) signal means the queue drained: resume. */
+        if (ece < (uint16_t) CA_ECE_REF)
+                ctx->ai_hold = false;
+
+        ctx->last_fb  = t;
+        ctx->last_sig = t;
         ctx->n_fb++;
 
         /* Scale the floor and AI slope to the path bottleneck. */
@@ -730,6 +839,44 @@ void mb_ecn_ctx_update_ece(void *   _ctx,
         mb_ecn_ece(ctx, ece, cap, TS_TO_UINT64(now));
 }
 
+/* Due when the path stayed quiet for a heartbeat interval; arms the gap. */
+bool mb_ecn_ctx_hb_due(void *   _ctx,
+                       uint64_t now)
+{
+        struct mb_ecn_ctx * ctx  = _ctx;
+        uint64_t            t_hb = mb_ecn_t_hb(ctx);
+        uint64_t            last;
+
+        last = ctx->last_sig > ctx->last_hb ? ctx->last_sig : ctx->last_hb;
+        if (now - last < t_hb)
+                return false;
+
+        ctx->last_hb = now;
+
+        return true;
+}
+
+/* Fold a heartbeat RTT sample into the ramp clock; also counts as life. */
+void mb_ecn_ctx_rtt(void *   _ctx,
+                    uint64_t now,
+                    uint64_t rtt)
+{
+        struct mb_ecn_ctx * ctx = _ctx;
+        uint64_t            tgt;
+
+        tgt = (uint64_t) CA_SS_RTT_MUL * rtt;
+        if (tgt < (uint64_t) CA_SS_TC_MIN)   /* track the true RTT both  */
+                tgt = CA_SS_TC_MIN;          /* ways: overshoot ~e^{1/2} */
+
+        if (tgt > (uint64_t) CA_SS_TC_MAX)   /* at the real RTT, not the */
+                tgt = CA_SS_TC_MAX;          /* declared worst case      */
+
+        ctx->ss_tc += (tgt >> CA_RTT_SHFT) - (ctx->ss_tc >> CA_RTT_SHFT);
+
+        ctx->last_sig = now;   /* liveness only: never ages the ece signal */
+        ctx->n_rtt++;
+}
+
 int  mb_ecn_calc_ecn(size_t    queued,
                      uint8_t * ecn,
                      qoscube_t qc,
@@ -762,7 +909,7 @@ ssize_t  mb_ecn_print_stats(void * _ctx,
         int                 code;
         uint16_t            m;
 
-        if (len < 1024)
+        if (len < CA_STATS_STRLEN)
                 return 0;
 
         /* No signal seen: the rate is unconstrained drift, not a target. */
@@ -776,6 +923,9 @@ ssize_t  mb_ecn_print_stats(void * _ctx,
         if (!ctx->tx_cav) {
                 regime = "Slow start";
                 code   = 0;
+        } else if (ctx->ai_hold) {
+                regime = "Loss recovery";
+                code   = 4;
         } else if (ctx->src_limited) {
                 regime = "Source limited";
                 code   = 3;
@@ -803,7 +953,10 @@ ssize_t  mb_ecn_print_stats(void * _ctx,
                 "Path capacity (bytes/s):         %20" PRIu64 "\n"
                 "Capacity rate floor (bytes/s):   %20" PRIu64 "\n"
                 "Capacity updates (count):        %20" PRIu64 "\n"
-                "Slow start peak rate (bytes/s):  %20" PRIu64 "\n",
+                "Slow start peak rate (bytes/s):  %20" PRIu64 "\n"
+                "Signal-loss cuts (count):        %20" PRIu64 "\n"
+                "Heartbeat RTT samples (count):   %20" PRIu64 "\n"
+                "Ramp time constant (ns):         %20" PRIu64 "\n",
                 "Multi-bit ECN",
                 ctx->tx_ece,
                 ctx->rx_ece,
@@ -812,7 +965,8 @@ ssize_t  mb_ecn_print_stats(void * _ctx,
                 ctx->n_ctrl, ctx->t_ctrl, ctx->t_bank,
                 ctx->n_fb, ctx->n_ttl,
                 cap_dec(ctx->tx_cap), ctx->rate_min, ctx->n_cap,
-                peak);
+                peak,
+                ctx->n_loss, ctx->n_rtt, ctx->ss_tc);
 
         return strlen(buf);
 }
diff --git a/src/ipcpd/unicast/ca/mb-ecn.h b/src/ipcpd/unicast/ca/mb-ecn.h
index 7bf0b29f..781e25a8 100644
--- a/src/ipcpd/unicast/ca/mb-ecn.h
+++ b/src/ipcpd/unicast/ca/mb-ecn.h
@@ -25,6 +25,8 @@
 
 #include "ops.h"
 
+void     mb_ecn_init(uint32_t rtt_ms);
+
 void *   mb_ecn_ctx_create(void);
 
 void     mb_ecn_ctx_destroy(void * ctx);
@@ -45,6 +47,13 @@ void     mb_ecn_ctx_update_ece(void *   ctx,
                                uint16_t ece,
                                uint8_t  cap);
 
+bool     mb_ecn_ctx_hb_due(void *   ctx,
+                           uint64_t now);
+
+void     mb_ecn_ctx_rtt(void *   ctx,
+                        uint64_t now,
+                        uint64_t rtt);
+
 int      mb_ecn_calc_ecn(size_t    queued,
                          uint8_t * ecn,
                          qoscube_t qc,
diff --git a/src/ipcpd/unicast/ca/ops.h b/src/ipcpd/unicast/ca/ops.h
index a86c1b3b..b01e8573 100644
--- a/src/ipcpd/unicast/ca/ops.h
+++ b/src/ipcpd/unicast/ca/ops.h
@@ -46,6 +46,14 @@ struct ca_ops {
                                     uint16_t ece,
                                     uint8_t  cap);
 
+        /* Optional, can be NULL: heartbeat pacing + RTT feedback. */
+        bool     (* ctx_hb_due)(void *   ctx,
+                                uint64_t now);
+
+        void     (* ctx_rtt)(void *   ctx,
+                             uint64_t now,
+                             uint64_t rtt);
+
         int      (* calc_ecn)(size_t    queued,
                               uint8_t * ecn,
                               qoscube_t qc,
diff --git a/src/ipcpd/unicast/ca/tests/ca_test.c b/src/ipcpd/unicast/ca/tests/ca_test.c
index 4c44d29e..1b86eab8 100644
--- a/src/ipcpd/unicast/ca/tests/ca_test.c
+++ b/src/ipcpd/unicast/ca/tests/ca_test.c
@@ -44,7 +44,7 @@ static int test_ca_init_fini(enum pol_cong_avoid pol,
 {
         TEST_START("(%s)", name);
 
-        if (ca_init(pol) < 0) {
+        if (ca_init(pol, 100) < 0) {
                 printf("Failed to init ca for %s.\n", name);
                 goto fail;
         }
@@ -74,7 +74,7 @@ static int test_ca_init_invalid(void)
 {
         TEST_START();
 
-        if (ca_init(CA_INVALID) == 0) {
+        if (ca_init(CA_INVALID, 100) == 0) {
                 printf("Init accepted an invalid policy.\n");
                 ca_fini();
                 goto fail;
@@ -96,7 +96,7 @@ static int test_ca_ctx_share(enum pol_cong_avoid pol,
 
         TEST_START("(%s)", name);
 
-        if (ca_init(pol) < 0) {
+        if (ca_init(pol, 100) < 0) {
                 printf("Failed to init ca for %s.\n", name);
                 goto fail;
         }
@@ -162,7 +162,7 @@ static int test_ca_ctx_distinct(void)
 
         TEST_START();
 
-        if (ca_init(CA_NONE) < 0) {
+        if (ca_init(CA_NONE, 100) < 0) {
                 printf("Failed to init ca.\n");
                 goto fail;
         }
@@ -226,7 +226,7 @@ static int test_ca_ctx_refcount(void)
 
         TEST_START();
 
-        if (ca_init(CA_NONE) < 0) {
+        if (ca_init(CA_NONE, 100) < 0) {
                 printf("Failed to init ca.\n");
                 goto fail;
         }
@@ -283,7 +283,7 @@ static int test_ca_ctx_recreate(void)
 
         TEST_START();
 
-        if (ca_init(CA_NONE) < 0) {
+        if (ca_init(CA_NONE, 100) < 0) {
                 printf("Failed to init ca.\n");
                 goto fail;
         }
@@ -340,7 +340,7 @@ static int test_ca_fini_drains(void)
 
         TEST_START();
 
-        if (ca_init(CA_NONE) < 0) {
+        if (ca_init(CA_NONE, 100) < 0) {
                 printf("Failed to init ca.\n");
                 goto fail;
         }
diff --git a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
index 8e3a73da..4bbc12aa 100644
--- a/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
+++ b/src/ipcpd/unicast/ca/tests/mb_ecn_test.c
@@ -21,7 +21,6 @@
  */
 
 #include "mb-ecn.c"
-
 #include <test/test.h>
 
 #define MS (MILLION) /* one millisecond in ns */
@@ -41,6 +40,7 @@ static struct mb_ecn_ctx * mk_ctx(void)
         ctx->last_ts   = 0;
         ctx->last_ctrl = 0;
         ctx->last_fb   = 0;
+        ctx->last_sig  = 0;
         ctx->last_loc  = 0;
         ctx->last_cap  = 0;
 
@@ -51,6 +51,7 @@ static struct mb_ecn_ctx * mk_ctx(void)
         ctx->backlogged  = true;
         ctx->src_limited = false;
         ctx->started     = false;
+        ctx->ss_tc       = 20 * MS; /* fixed slope for deterministic SS */
 
         return ctx;
 }
@@ -815,7 +816,7 @@ static int test_mb_ecn_rcv_no_overflow_highrate(void)
 
 /*
  * The sender holds a mark across the full inter-feedback gap (TTL >
- * 2 * CA_TW_INIT) and a repeated mark must not re-fire the one-sided lead.
+ * 2 * CA_TW_INIT); a repeated mark adds only the proportional cut.
  */
 static int test_mb_ecn_ece_ttl_covers_cadence(void)
 {
@@ -1048,10 +1049,13 @@ static int test_mb_ecn_rate_floor(void)
                 goto fail;
         }
 
-        /* Cut larger than headroom must clamp; stay inside CA_ECE_TTL. */
+        /*
+         * A starved 600 ms at full mark takes the rate/2 branch, far
+         * below the floor; stays inside the initial ~976 ms mark TTL.
+         */
         ctx->rate = CA_RATE_MIN + 1000;
         mb_ecn_ece(ctx, CA_ECE_REF, 0, 0);
-        mb_ecn_snd(ctx, LEN, 30 * MS, &ftag);
+        mb_ecn_snd(ctx, LEN, 600 * MS, &ftag);
 
         if (ctx->rate != CA_RATE_MIN) {
                 printf("rate floor breached: %" PRIu64 ".\n", ctx->rate);
@@ -1116,8 +1120,9 @@ static int test_mb_ecn_fixed_point(void)
 }
 
 /*
- * One-sided lead: a mark rise cuts rate * rise / (REF * DEN) once;
- * a flat or falling mark leaves only the dt-scaled proportional cut.
+ * A rising ECE mark takes a one-sided lead cut on top of the
+ * proportional term; a flat or falling mark takes only the small
+ * proportional cut.
  */
 static int test_mb_ecn_lead_cut(void)
 {
@@ -1137,17 +1142,17 @@ static int test_mb_ecn_lead_cut(void)
         ctx->rate = (uint64_t) 100 << 20;
         ctx->tx_cav = true;
 
-        /* Rise 0 -> 256: lead cuts ~rise/(REF*DEN) = 1/8 of the rate. */
+        /* Rise 0 -> 256: the lead term cuts hard (~rate/8). */
         prev = ctx->rate;
         mb_ecn_ece(ctx, 256, 0, MS);
         mb_ecn_snd(ctx, LEN, MS, &ftag);
         drop = prev - ctx->rate;
-        if (drop < prev / 10) {
-                printf("lead cut missing: dropped %" PRIu64 ".\n", drop);
+        if (drop < prev / 16) {
+                printf("rising mark under-cut: %" PRIu64 ".\n", drop);
                 goto fail_ctx;
         }
 
-        /* Flat mark: rise 0, only the ~0.05%% proportional cut. */
+        /* Flat mark: no rise, only the proportional cut. */
         prev = ctx->rate;
         mb_ecn_ece(ctx, 256, 0, 2 * MS);
         mb_ecn_snd(ctx, LEN, 2 * MS, &ftag);
@@ -1157,7 +1162,7 @@ static int test_mb_ecn_lead_cut(void)
                 goto fail_ctx;
         }
 
-        /* Falling mark: one-sided lead must not fire. */
+        /* Falling mark: no rise, no lead beyond the proportional term. */
         prev = ctx->rate;
         mb_ecn_ece(ctx, 64, 0, 3 * MS);
         mb_ecn_snd(ctx, LEN, 3 * MS, &ftag);
@@ -1421,6 +1426,69 @@ static int test_mb_ecn_decrease_dt_invariant(void)
         return TEST_RC_FAIL;
 }
 
+/* Sub-ms control steps must not lose decrease time to truncation. */
+static int test_mb_ecn_decrease_subms_carry(void)
+{
+        struct mb_ecn_ctx * a;
+        struct mb_ecn_ctx * b;
+        uint64_t            cut_a;
+        uint64_t            cut_b;
+        uint64_t            r0;
+        uint64_t            fta = 0;
+        uint64_t            ftb = 0;
+        size_t              i;
+
+        TEST_START();
+
+        a = mk_ctx();
+        b = mk_ctx();
+        if (a == NULL || b == NULL) {
+                printf("Failed to create contexts.\n");
+                goto fail_ctx;
+        }
+
+        r0 = (uint64_t) 100 << 20;
+        a->rate = r0;
+        b->rate = r0;
+
+        mb_ecn_ece(a, 256, 0, 0);
+        mb_ecn_ece(b, 256, 0, 0);
+
+        /* Same 30 ms of marked time; b's steps have a 0.5 ms tail. */
+        for (i = 1; i <= 10; i++)
+                mb_ecn_snd(a, LEN, i * 3 * MS, &fta);
+
+        for (i = 1; i <= 20; i++)
+                mb_ecn_snd(b, LEN, i * 3 * MS / 2, &ftb);
+
+        cut_a = r0 - a->rate;
+        cut_b = r0 - b->rate;
+
+        if (cut_a == 0 || cut_b == 0) {
+                printf("no cut: %" PRIu64 " %" PRIu64 ".\n", cut_a, cut_b);
+                goto fail_ctx;
+        }
+
+        /* Within 10%: the sub-ms remainder must carry, not vanish. */
+        if (cut_a > cut_b + cut_a / 10 || cut_b > cut_a + cut_a / 10) {
+                printf("sub-ms decrease lost: %" PRIu64 " vs %" PRIu64
+                       ".\n", cut_a, cut_b);
+                goto fail_ctx;
+        }
+
+        mb_ecn_ctx_destroy(a);
+        mb_ecn_ctx_destroy(b);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail_ctx:
+        mb_ecn_ctx_destroy(a);
+        mb_ecn_ctx_destroy(b);
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
 /* Resuming after a long idle gap: bounded AI, no cut from stale marks. */
 static int test_mb_ecn_idle_resume_bounded(void)
 {
@@ -2528,6 +2596,269 @@ static int test_mb_ecn_first_send_warmup(void)
         return TEST_RC_FAIL;
 }
 
+/*
+ * Couple one or two backlogged flows through a shared bottleneck of
+ * capacity cap: each step marks the shared queue with the real
+ * calc_ecn, feeds each flow that mark delayed by its own lag (in
+ * steps), drives it backlogged for one step, then drains the queue.
+ * A faithful discrete run of the fluid model on the real control law,
+ * with the forwarder abstracted to a single shared price. b may be
+ * NULL for a single-flow run.
+ */
+#define TF_STEP  (5 * MS)          /* control step (ns)         */
+#define TF_HIST  64                /* mark ring, bounds max lag */
+#define TF_QMAX  (8192 * LEN)      /* bottleneck buffer (bytes) */
+#define TF_MAXN  32                /* flows per shared-link run */
+
+static void shared_link_run(struct mb_ecn_ctx * a,
+                            struct mb_ecn_ctx * b,
+                            uint64_t            cap,
+                            size_t              lag_a,
+                            size_t              lag_b,
+                            size_t              steps)
+{
+        uint16_t hist[TF_HIST];
+        uint64_t ta    = 0;
+        uint64_t tb    = 0;
+        uint64_t fta   = 0;
+        uint64_t ftb   = 0;
+        uint64_t q     = 0;
+        uint64_t drain = cap * TF_STEP / BILLION;
+        uint64_t tgt;
+        uint64_t arr;
+        uint8_t  cc    = cap_enc(cap);
+        uint8_t  ecn;
+        uint16_t ea;
+        uint16_t eb;
+        size_t   k;
+
+        memset(hist, 0, sizeof(hist));
+
+        for (k = 0; k < steps; k++) {
+                tgt = (k + 1) * TF_STEP;
+                ecn = 0;
+
+                mb_ecn_calc_ecn(q / LEN, &ecn, QOS_CUBE_BE, LEN);
+                hist[k % TF_HIST] = (uint16_t) (ecn << CA_SHFT);
+
+                ea = k < lag_a ? 0 : hist[(k - lag_a) % TF_HIST];
+                if (ea > 0)         /* no feedback until congestion */
+                        mb_ecn_ece(a, ea, cc, ta);
+
+                /* Active flow: heartbeat/window liveness stays fresh. */
+                a->last_sig = ta;
+                if (tgt > ta)
+                        ta = drive_backlogged(a, &fta, ta, tgt - ta, LEN);
+
+                arr = a->rate * TF_STEP / BILLION;
+
+                if (b != NULL) {
+                        eb = k < lag_b ? 0 : hist[(k - lag_b) % TF_HIST];
+                        if (eb > 0)
+                                mb_ecn_ece(b, eb, cc, tb);
+
+                        b->last_sig = tb;
+                        if (tgt > tb)
+                                tb = drive_backlogged(b, &ftb, tb,
+                                                      tgt - tb, LEN);
+
+                        arr += b->rate * TF_STEP / BILLION;
+                }
+
+                q += arr;
+                q  = q > drain ? q - drain : 0;
+                if (q > (uint64_t) TF_QMAX)
+                        q = TF_QMAX;
+        }
+}
+
+/*
+ * Two flows sharing one bottleneck, both fed every queue mark, must
+ * converge from a lopsided start toward an equal split (the fluid
+ * model's drho/dt -> 0) rather than latch winner-take-all. Isolates
+ * the rate law from the forwarder: neither flow is starved of marks.
+ */
+static int test_mb_ecn_two_flow_converge(void)
+{
+        struct mb_ecn_ctx * a;
+        struct mb_ecn_ctx * b;
+        uint64_t            cap  = 1ULL << 20;
+        uint64_t            fair = (1ULL << 20) / 2;
+        uint64_t            lo;
+        uint64_t            hi;
+
+        TEST_START();
+
+        a = mk_ctx();
+        b = mk_ctx();
+        if (a == NULL || b == NULL) {
+                printf("Failed to create contexts.\n");
+                goto fail_ctx;
+        }
+
+        a->rate = cap;             /* a hogs, b starts small */
+        b->rate = CA_RATE_INIT;
+
+        shared_link_run(a, b, cap, 0, 0, 6000);
+
+        lo = a->rate < b->rate ? a->rate : b->rate;
+        hi = a->rate > b->rate ? a->rate : b->rate;
+
+        if (lo < fair / 4) {
+                printf("flow starved: %" PRIu64 " / %" PRIu64 ".\n",
+                       a->rate, b->rate);
+                goto fail_ctx;
+        }
+
+        if (hi > 3 * lo) {
+                printf("did not converge: %" PRIu64 " / %" PRIu64 ".\n",
+                       a->rate, b->rate);
+                goto fail_ctx;
+        }
+
+        mb_ecn_ctx_destroy(a);
+        mb_ecn_ctx_destroy(b);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail_ctx:
+        mb_ecn_ctx_destroy(a);
+        mb_ecn_ctx_destroy(b);
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/*
+ * A remote bottleneck exits slow start only on fed-back congestion,
+ * so a longer feedback delay lets the exponential ramp overshoot
+ * further: the slow-start peak grows with RTT. Nothing caps the ramp
+ * at the path capacity, so the sender latches near the first-hop rate.
+ */
+static int test_mb_ecn_ramp_overshoot_grows_with_rtt(void)
+{
+        struct mb_ecn_ctx * lo_rtt;
+        struct mb_ecn_ctx * hi_rtt;
+        uint64_t            cap = 1ULL << 20;
+
+        TEST_START();
+
+        lo_rtt = mk_ctx();
+        hi_rtt = mk_ctx();
+        if (lo_rtt == NULL || hi_rtt == NULL) {
+                printf("Failed to create contexts.\n");
+                goto fail_ctx;
+        }
+
+        /* Same bottleneck; feedback lags 1 vs 8 steps (~5 vs 40 ms). */
+        shared_link_run(lo_rtt, NULL, cap, 1, 0, 2000);
+        shared_link_run(hi_rtt, NULL, cap, 8, 0, 2000);
+
+        if (hi_rtt->ss_peak <= lo_rtt->ss_peak) {
+                printf("overshoot did not grow with RTT: %" PRIu64
+                       " vs %" PRIu64 ".\n",
+                       hi_rtt->ss_peak, lo_rtt->ss_peak);
+                goto fail_ctx;
+        }
+
+        mb_ecn_ctx_destroy(lo_rtt);
+        mb_ecn_ctx_destroy(hi_rtt);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail_ctx:
+        mb_ecn_ctx_destroy(lo_rtt);
+        mb_ecn_ctx_destroy(hi_rtt);
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/*
+ * Washout damps a fixed 1/4 of the rate change once per wall-clock
+ * bucket: a sub-bucket step banks time only, a full bucket removes a
+ * quarter of the gap either way without crossing the snapshot, and a
+ * sparse step (> CA_DT_CAP) resets it so a starved sender keeps its cut.
+ */
+static int test_mb_ecn_washout_bucket(void)
+{
+        struct mb_ecn_ctx * ctx;
+        uint64_t            r0 = (uint64_t) 100 << 20;
+        uint64_t            gap;
+
+        TEST_START();
+
+        ctx = mk_ctx();
+        if (ctx == NULL) {
+                printf("Failed to create context.\n");
+                goto fail;
+        }
+
+        gap = r0 >> 4;
+
+        /* Sub-bucket: time banks, the rate does not move. */
+        ctx->rate     = r0;
+        ctx->r_bkt    = r0 - gap;
+        ctx->wash_acc = 0;
+        mb_ecn_washout(ctx, MS, MS);
+        if (ctx->rate != r0 || ctx->wash_acc != MS) {
+                printf("sub-bucket washout moved the rate: %" PRIu64 ".\n",
+                       ctx->rate);
+                goto fail_ctx;
+        }
+
+        /* Full bucket, rate leads: cut a quarter of the gap, no crossing. */
+        ctx->rate     = r0;
+        ctx->r_bkt    = r0 - gap;
+        ctx->wash_acc = 0;
+        mb_ecn_washout(ctx, MS, CA_WASH_BKT);
+        if (ctx->rate != r0 - (gap >> CA_WASH_SHFT)) {
+                printf("bucket down: got %" PRIu64 ".\n", ctx->rate);
+                goto fail_ctx;
+        }
+
+        if (ctx->rate <= r0 - gap) {
+                printf("washout reversed a ramp: %" PRIu64 ".\n", ctx->rate);
+                goto fail_ctx;
+        }
+
+        if (ctx->r_bkt != ctx->rate || ctx->wash_acc != 0) {
+                printf("washout did not reset the bucket.\n");
+                goto fail_ctx;
+        }
+
+        /* Full bucket, rate trails: add a quarter of the gap (symmetric). */
+        ctx->rate     = r0;
+        ctx->r_bkt    = r0 + gap;
+        ctx->wash_acc = 0;
+        mb_ecn_washout(ctx, MS, CA_WASH_BKT);
+        if (ctx->rate != r0 + (gap >> CA_WASH_SHFT)) {
+                printf("bucket up: got %" PRIu64 ".\n", ctx->rate);
+                goto fail_ctx;
+        }
+
+        /* Sparse step: reset to the current rate, keep the cut. */
+        ctx->rate     = r0;
+        ctx->r_bkt    = r0 - gap;
+        ctx->wash_acc = CA_WASH_BKT / 2;
+        mb_ecn_washout(ctx, CA_DT_CAP + 1, MS);
+        if (ctx->rate != r0 || ctx->r_bkt != r0 || ctx->wash_acc != 0) {
+                printf("sparse step did not reset the bucket.\n");
+                goto fail_ctx;
+        }
+
+        mb_ecn_ctx_destroy(ctx);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail_ctx:
+        mb_ecn_ctx_destroy(ctx);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
 int mb_ecn_test(int     argc,
                 char ** argv)
 {
@@ -2561,6 +2892,7 @@ int mb_ecn_test(int     argc,
         ret |= test_mb_ecn_starved_decrease_escape();
         ret |= test_mb_ecn_starved_local_fallback();
         ret |= test_mb_ecn_decrease_dt_invariant();
+        ret |= test_mb_ecn_decrease_subms_carry();
         ret |= test_mb_ecn_idle_resume_bounded();
         ret |= test_mb_ecn_rate_floor();
         ret |= test_mb_ecn_ece_staleness();
@@ -2583,6 +2915,9 @@ int mb_ecn_test(int     argc,
         ret |= test_mb_ecn_max_filter();
         ret |= test_mb_ecn_idle_clears_backlogged();
         ret |= test_mb_ecn_first_send_warmup();
+        ret |= test_mb_ecn_two_flow_converge();
+        ret |= test_mb_ecn_ramp_overshoot_grows_with_rtt();
+        ret |= test_mb_ecn_washout_bucket();
 
         return ret;
 }
diff --git a/src/ipcpd/unicast/dt.c b/src/ipcpd/unicast/dt.c
index 381ed815..bfc2ece7 100644
--- a/src/ipcpd/unicast/dt.c
+++ b/src/ipcpd/unicast/dt.c
@@ -595,7 +595,7 @@ int dt_init(struct dt_config cfg)
                 goto fail_cap;
         }
 
-        if (connmgr_comp_init(COMPID_DT, &info)) {
+        if (connmgr_comp_init(COMPID_DT, &info) != 0) {
                 log_err("Failed to register with connmgr.");
                 goto fail_connmgr_comp_init;
         }
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index ac2ecaea..596b101b 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -32,17 +32,18 @@
 #define OUROBOROS_PREFIX FA
 
 #include <ouroboros/atomics.h>
+#include <ouroboros/dev.h>
 #include <ouroboros/endian.h>
-#include <ouroboros/logs.h>
-#include <ouroboros/fqueue.h>
 #include <ouroboros/errno.h>
-#include <ouroboros/dev.h>
+#include <ouroboros/fqueue.h>
 #include <ouroboros/ipcp-dev.h>
+#include <ouroboros/logs.h>
 #include <ouroboros/np1_flow.h>
+#include <ouroboros/pthread.h>
 #include <ouroboros/qoscube.h>
-#include <ouroboros/rib.h>
 #include <ouroboros/random.h>
-#include <ouroboros/pthread.h>
+#include <ouroboros/rib.h>
+#include <ouroboros/time.h>
 
 #include "addr-auth.h"
 #include "dir.h"
@@ -68,6 +69,10 @@
 #define FLOW_REPLY      1
 #define FLOW_UPDATE     2
 #define FLOW_IRM_UPDATE 3
+#define FLOW_HB         4
+#define FLOW_ACK        5
+
+#define HB_ID_LEN       16   /* 128-bit unguessable heartbeat nonce   */
 
 #define STAT_FILE_LEN 0
 
@@ -87,6 +92,7 @@ struct fa_msg {
         uint8_t  code;
         uint8_t  availability;
         uint8_t  service;
+        uint8_t  hb_id[HB_ID_LEN]; /* heartbeat / ack nonce           */
 } __attribute__((packed));
 
 struct cmd {
@@ -94,6 +100,22 @@ struct cmd {
         struct ssm_pk_buff * spb;
 };
 
+#define HB_TBL_TTL (4ULL * BILLION)  /* drop unanswered heartbeats  */
+#define HB_TBL_MAX 1024              /* cap outstanding heartbeats  */
+#define HB_BUCKETS 256               /* nonce hash buckets (pow2)   */
+
+/* RIB flow entry: the CA stats string plus the flow header. */
+#define FA_RIB_STRLEN (CA_STATS_STRLEN + 512)
+
+/* Outstanding heartbeat: send time kept locally, keyed by the nonce. */
+struct hb_ent {
+        uint8_t          id[HB_ID_LEN];
+        uint64_t         s_eid;    /* originating flow (fd-reuse guard) */
+        uint64_t         t_snd;    /* send timestamp (ns)               */
+        struct list_head hnext;    /* nonce hash bucket chain           */
+        struct list_head qnext;    /* expiry FIFO, oldest at head       */
+};
+
 struct fa_flow {
 #ifdef IPCP_FLOW_STATS
         time_t   stamp;    /* Flow creation                  */
@@ -129,6 +151,11 @@ struct {
         pthread_mutex_t  mtx;
         pthread_t        worker;
 
+        struct list_head hb_bkt[HB_BUCKETS]; /* nonce hash buckets    */
+        struct list_head hb_q;   /* expiry FIFO, oldest at head       */
+        size_t           n_hbs;
+        pthread_mutex_t  hb_mtx;
+
         struct psched *  psched;
 } fa;
 
@@ -143,7 +170,7 @@ static int fa_rib_read(const char * path,
         char             s_eidstr[21];
         char             r_eidstr[21];
         char             tmstr[RIB_TM_STRLEN];
-        char             castr[1024];
+        char             castr[CA_STATS_STRLEN];
         char *           entry;
         struct tm *      tm;
 
@@ -155,7 +182,7 @@ static int fa_rib_read(const char * path,
         if (fd < 0 || fd >= PROC_MAX_FLOWS)
                 return -1;
 
-        if (len < 1536)
+        if (len < FA_RIB_STRLEN)
                 return 0;
 
         flow = &fa.flows[fd];
@@ -176,7 +203,7 @@ static int fa_rib_read(const char * path,
         tm = gmtime(&flow->stamp);
         strftime(tmstr, sizeof(tmstr), RIB_TM_FORMAT, tm);
 
-        ca_print_stats(flow->ctx, castr, 1024);
+        ca_print_stats(flow->ctx, castr, CA_STATS_STRLEN);
 
         sprintf(buf,
                 "Flow established at:             %20s\n"
@@ -274,7 +301,7 @@ static int fa_rib_getattr(const char *      path,
         pthread_rwlock_rdlock(&fa.flows_lock);
 
         if (flow->stamp != 0) {
-                attr->size  = 1536;
+                attr->size  = FA_RIB_STRLEN;
                 attr->mtime = flow->stamp;
         } else {
                 attr->size  = 0;
@@ -323,13 +350,129 @@ static uint64_t gen_eid(int fd)
         return ((uint64_t) rnd << 32) + fd;
 }
 
+/* The nonce is uniformly random, so its low word is a fine hash. */
+static size_t fa_hb_hash(const uint8_t * id)
+{
+        uint32_t h;
+
+        memcpy(&h, id, sizeof(h));
+
+        return h & (HB_BUCKETS - 1);
+}
+
+/* Record an outstanding heartbeat; expire stale entries as we go. */
+static void fa_hb_record(const uint8_t * id,
+                         uint64_t        s_eid,
+                         uint64_t        t_snd)
+{
+        struct hb_ent *    ent;
+        struct list_head * p;
+        struct list_head * h;
+
+        ent = malloc(sizeof(*ent));
+        if (ent == NULL)
+                return;
+
+        memcpy(ent->id, id, HB_ID_LEN);
+        ent->s_eid = s_eid;
+        ent->t_snd = t_snd;
+
+        pthread_mutex_lock(&fa.hb_mtx);
+
+        /* The FIFO is time-ordered; stop at the first fresh entry. */
+        list_for_each_safe(p, h, &fa.hb_q) {
+                struct hb_ent * e = list_entry(p, struct hb_ent, qnext);
+                if (t_snd - e->t_snd <= HB_TBL_TTL)
+                        break;
+                list_del(&e->hnext);
+                list_del(&e->qnext);
+                free(e);
+                fa.n_hbs--;
+        }
+
+        if (fa.n_hbs >= HB_TBL_MAX) {
+                pthread_mutex_unlock(&fa.hb_mtx);
+                free(ent);
+                return;
+        }
+
+        list_add(&ent->hnext, &fa.hb_bkt[fa_hb_hash(id)]);
+        list_add_tail(&ent->qnext, &fa.hb_q);
+        fa.n_hbs++;
+
+        pthread_mutex_unlock(&fa.hb_mtx);
+}
+
+/* Consume a heartbeat nonce, returning the flow and send time it maps to. */
+static int fa_hb_match(const uint8_t * id,
+                       uint64_t *      s_eid,
+                       uint64_t *      t_snd)
+{
+        struct list_head * bkt;
+        struct list_head * p;
+        struct list_head * h;
+
+        pthread_mutex_lock(&fa.hb_mtx);
+
+        bkt = &fa.hb_bkt[fa_hb_hash(id)];
+        list_for_each_safe(p, h, bkt) {
+                struct hb_ent * e = list_entry(p, struct hb_ent, hnext);
+                if (memcmp(e->id, id, HB_ID_LEN) == 0) {
+                        *s_eid = e->s_eid;
+                        *t_snd = e->t_snd;
+                        list_del(&e->hnext);
+                        list_del(&e->qnext);
+                        free(e);
+                        fa.n_hbs--;
+                        pthread_mutex_unlock(&fa.hb_mtx);
+                        return 0;
+                }
+        }
+
+        pthread_mutex_unlock(&fa.hb_mtx);
+
+        return -1;
+}
+
+/* Send a bare control message (heartbeat or ack) carrying only a nonce. */
+static int fa_send_ctrl(uint64_t        r_addr,
+                        uint8_t         code,
+                        const uint8_t * id)
+{
+        struct fa_msg *      msg;
+        struct ssm_pk_buff * spb;
+        qoscube_t            qc = QOS_CUBE_BE;
+
+        if (ipcp_spb_reserve(&spb, sizeof(*msg)))
+                return -1;
+
+        msg = (struct fa_msg *) ssm_pk_buff_head(spb);
+        memset(msg, 0, sizeof(*msg));
+
+        msg->code   = code;
+        msg->s_addr = hton64(addr_auth_address());
+        memcpy(msg->hb_id, id, HB_ID_LEN);
+
+        if (dt_write_packet(r_addr, qc, fa.eid, spb, NULL)) {
+                ipcp_spb_release(spb);
+                return -1;
+        }
+
+        return 0;
+}
+
 static time_t packet_handler(int                  fd,
                              qoscube_t            qc,
                              struct ssm_pk_buff * spb)
 {
         struct fa_flow * flow;
+        struct timespec  tv;
+        uint64_t         now;
         uint64_t         r_addr;
         uint64_t         r_eid;
+        uint64_t         s_eid;
+        bool             hb;
+        uint8_t          nonce[HB_ID_LEN];
         time_t           wait;
         size_t           len;
         uint8_t          ecn;
@@ -338,6 +481,9 @@ static time_t packet_handler(int                  fd,
 
         ecn = 0;
 
+        clock_gettime(PTHREAD_COND_CLOCK, &tv);
+        now = TS_TO_UINT64(tv);
+
         pthread_rwlock_wrlock(&fa.flows_lock);
 
         len = ssm_pk_buff_len(spb);
@@ -349,11 +495,18 @@ static time_t packet_handler(int                  fd,
         wait = ca_ctx_update_snd(flow->ctx, len,
                                  LOAD_RELAXED(&flow->l_ecn), &flow->fair);
 
+        hb     = ca_ctx_hb_due(flow->ctx, now);
         r_addr = flow->r_addr;
         r_eid  = flow->r_eid;
+        s_eid  = flow->s_eid;
 
         pthread_rwlock_unlock(&fa.flows_lock);
 
+        if (hb && random_buffer(nonce, HB_ID_LEN) == 0) {
+                fa_hb_record(nonce, s_eid, now);
+                fa_send_ctrl(r_addr, FLOW_HB, nonce);
+        }
+
         if (dt_write_packet(r_addr, qc, r_eid, spb, &ecn)) {
                 STORE_RELAXED(&flow->l_ecn, ecn);
                 ipcp_spb_release(spb);
@@ -500,6 +653,9 @@ static int fa_handle_flow_req(struct fa_msg * msg,
         qs.max_gap      = ntoh32(msg->max_gap);
         qs.timeout      = ntoh32(msg->timeout);
 
+        /* Ack the seed nonce now: a clean RTT, before any accept delay. */
+        fa_send_ctrl(ntoh64(msg->s_addr), FLOW_ACK, msg->hb_id);
+
         fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_UNICAST_MPL,
                                     IPCP_UNICAST_MTU, &data);
         if (fd < 0)
@@ -606,6 +762,49 @@ static int fa_handle_flow_update(struct fa_msg * msg,
         return 0;
 }
 
+/* Heartbeat: reflect the nonce straight back to the sender's address. */
+static int fa_handle_flow_hb(struct fa_msg * msg,
+                             size_t          len)
+{
+        if (len < sizeof(*msg))
+                return -EINVAL;
+
+        return fa_send_ctrl(ntoh64(msg->s_addr), FLOW_ACK, msg->hb_id);
+}
+
+/* Ack: the reflected nonce yields an RTT sample for its path. */
+static int fa_handle_flow_ack(struct fa_msg * msg,
+                              size_t          len)
+{
+        struct timespec  tv;
+        struct fa_flow * flow;
+        uint64_t         now;
+        uint64_t         t_snd;
+        uint64_t         s_eid;
+        int              fd;
+
+        if (len < sizeof(*msg))
+                return -EINVAL;
+
+        if (fa_hb_match(msg->hb_id, &s_eid, &t_snd) < 0)
+                return 0;  /* unknown or stale nonce */
+
+        clock_gettime(PTHREAD_COND_CLOCK, &tv);
+        now = TS_TO_UINT64(tv);
+
+        pthread_rwlock_wrlock(&fa.flows_lock);
+
+        fd = eid_to_fd(s_eid);
+        if (fd >= 0 && now > t_snd) {
+                flow = &fa.flows[fd];
+                ca_ctx_rtt(flow->ctx, now, now - t_snd);
+        }
+
+        pthread_rwlock_unlock(&fa.flows_lock);
+
+        return 0;
+}
+
 static int fa_handle_flow_irm_update(struct fa_msg * msg,
                                size_t          len)
 {
@@ -675,6 +874,14 @@ static void * fa_handle_packet(void * o)
                         if (fa_handle_flow_irm_update(msg, len) < 0)
                                 log_err("Error handling flow update.");
                         break;
+                case FLOW_HB:
+                        if (fa_handle_flow_hb(msg, len) < 0)
+                                log_err("Error handling heartbeat.");
+                        break;
+                case FLOW_ACK:
+                        if (fa_handle_flow_ack(msg, len) < 0)
+                                log_err("Error handling heartbeat ack.");
+                        break;
                 default:
                         log_warn("Recieved unknown flow allocation message.");
                         break;
@@ -687,24 +894,28 @@ static void * fa_handle_packet(void * o)
 int fa_init(void)
 {
         pthread_condattr_t cattr;
+        size_t             i;
 
-        if (pthread_rwlock_init(&fa.flows_lock, NULL))
+        if (pthread_rwlock_init(&fa.flows_lock, NULL) != 0)
                 goto fail_rwlock;
 
-        if (pthread_mutex_init(&fa.mtx, NULL))
+        if (pthread_mutex_init(&fa.mtx, NULL) != 0)
                 goto fail_mtx;
 
-        if (pthread_condattr_init(&cattr))
+        if (pthread_mutex_init(&fa.hb_mtx, NULL) != 0)
+                goto fail_hb_mtx;
+
+        if (pthread_condattr_init(&cattr) != 0)
                 goto fail_cattr;
 
 #ifndef __APPLE__
         pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
 #endif
-        if (pthread_cond_init(&fa.cond, &cattr))
+        if (pthread_cond_init(&fa.cond, &cattr) != 0)
                 goto fail_cond;
 
 #ifdef IPCP_FLOW_STATS
-        if (rib_reg(FA, &r_ops))
+        if (rib_reg(FA, &r_ops) != 0)
                 goto fail_rib_reg;
 #endif
 
@@ -713,6 +924,11 @@ int fa_init(void)
                 goto fail_dt_reg;
 
         list_head_init(&fa.cmds);
+        for (i = 0; i < HB_BUCKETS; i++)
+                list_head_init(&fa.hb_bkt[i]);
+
+        list_head_init(&fa.hb_q);
+        fa.n_hbs = 0;
 
         pthread_condattr_destroy(&cattr);
 
@@ -727,6 +943,8 @@ int fa_init(void)
  fail_cond:
         pthread_condattr_destroy(&cattr);
  fail_cattr:
+        pthread_mutex_destroy(&fa.hb_mtx);
+ fail_hb_mtx:
         pthread_mutex_destroy(&fa.mtx);
  fail_mtx:
         pthread_rwlock_destroy(&fa.flows_lock);
@@ -736,10 +954,20 @@ int fa_init(void)
 
 void fa_fini(void)
 {
+        struct list_head * p;
+        struct list_head * h;
+
 #ifdef IPCP_FLOW_STATS
         rib_unreg(FA);
 #endif
+        list_for_each_safe(p, h, &fa.hb_q) {
+                struct hb_ent * e = list_entry(p, struct hb_ent, qnext);
+                list_del(&e->qnext);
+                free(e);
+        }
+
         pthread_cond_destroy(&fa.cond);;
+        pthread_mutex_destroy(&fa.hb_mtx);
         pthread_mutex_destroy(&fa.mtx);
         pthread_rwlock_destroy(&fa.flows_lock);
 }
@@ -822,6 +1050,8 @@ int fa_alloc(int              fd,
         qoscube_t            qc = QOS_CUBE_BE;
         size_t               len;
         uint64_t             eid;
+        struct timespec      tv;
+        uint8_t              nonce[HB_ID_LEN];
 
         addr = dir_query(dst);
         if (addr == 0)
@@ -849,6 +1079,13 @@ int fa_alloc(int              fd,
         msg->max_gap      = hton32(qs.max_gap);
         msg->timeout      = hton32(qs.timeout);
 
+        /* Seed an early RTT sample: peer acks this nonce on arrival. */
+        if (random_buffer(nonce, HB_ID_LEN) == 0) {
+                memcpy(msg->hb_id, nonce, HB_ID_LEN);
+                clock_gettime(PTHREAD_COND_CLOCK, &tv);
+                fa_hb_record(nonce, eid, TS_TO_UINT64(tv));
+        }
+
         memcpy(msg + 1, dst, ipcp_dir_hash_len());
         if (data->len > 0)
                 memcpy(ssm_pk_buff_head(spb) + len, data->data, data->len);
diff --git a/src/ipcpd/unicast/main.c b/src/ipcpd/unicast/main.c
index 1155b88b..86cb1f06 100644
--- a/src/ipcpd/unicast/main.c
+++ b/src/ipcpd/unicast/main.c
@@ -67,7 +67,7 @@ static int initialize_components(struct ipcp_config * conf)
 
         log_info("IPCP got address %" PRIu64 ".", addr_auth_address());
 
-        if (ca_init(conf->unicast.cong_avoid)) {
+        if (ca_init(conf->unicast.cong_avoid, conf->unicast.dt.max_rtt)) {
                 log_err("Failed to initialize congestion avoidance.");
                 goto fail_ca;
         }
diff --git a/src/irmd/configfile.c b/src/irmd/configfile.c
index 5967e490..e2e1e554 100644
--- a/src/irmd/configfile.c
+++ b/src/irmd/configfile.c
@@ -477,6 +477,7 @@ static int toml_dt(toml_table_t *     table,
         toml_datum_t addr;
         toml_datum_t eid;
         toml_datum_t ttl;
+        toml_datum_t max_rtt;
 
         addr = toml_int_in(table, "addr_size");
         if (addr.ok)
@@ -490,6 +491,10 @@ static int toml_dt(toml_table_t *     table,
         if (ttl.ok)
                 conf->max_ttl = ttl.u.i;
 
+        max_rtt = toml_int_in(table, "max_rtt");
+        if (max_rtt.ok)
+                conf->max_rtt = max_rtt.u.i;
+
         if (toml_routing(table, conf) < 0) {
                 log_err("Invalid routing option.");
                 return -1;
diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto
index eac4da37..a111b6d8 100644
--- a/src/lib/pb/ipcp_config.proto
+++ b/src/lib/pb/ipcp_config.proto
@@ -42,6 +42,7 @@ message dt_config_msg {
         required uint32 eid_size            = 2;
         required uint32 max_ttl             = 3;
         required routing_config_msg routing = 4;
+        optional uint32 max_rtt             = 5;
 }
 
 message dir_dht_config_msg {
diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c
index a824d357..6beae000 100644
--- a/src/lib/protobuf.c
+++ b/src/lib/protobuf.c
@@ -362,6 +362,8 @@ dt_config_msg_t * dt_config_s_to_msg(const struct dt_config * s)
         msg->addr_size    = s->addr_size;
         msg->eid_size     = s->eid_size;
         msg->max_ttl      = s->max_ttl;
+        msg->has_max_rtt  = true;
+        msg->max_rtt      = s->max_rtt;
         msg->routing      = routing_config_s_to_msg(&s->routing);
         if (msg->routing == NULL)
                 goto fail_routing;
@@ -381,6 +383,7 @@ struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg)
         s.addr_size    = msg->addr_size;
         s.eid_size     = msg->eid_size;
         s.max_ttl      = msg->max_ttl;
+        s.max_rtt      = msg->has_max_rtt ? msg->max_rtt : 200; /* ms       */
         s.routing      = routing_config_msg_to_s(msg->routing);
 
         return s;
diff --git a/src/tools/irm/irm_ipcp_bootstrap.c b/src/tools/irm/irm_ipcp_bootstrap.c
index de73b076..cc8bf8fa 100644
--- a/src/tools/irm/irm_ipcp_bootstrap.c
+++ b/src/tools/irm/irm_ipcp_bootstrap.c
@@ -105,7 +105,8 @@ static char * usage_str = \
         "    [Data Transfer Constants]\n"
         "        [addr <address size> (default: %d)]\n"
         "        [eid <eid size> (default: %d)]\n"
-        "        [ttl <max time-to-live>, default: %d)]\n\n"
+        "        [ttl <max time-to-live>, default: %d)]\n"
+        "        [rtt <max layer RTT in ms> (default: %d)]\n\n"
         "if DIRECTORY_POLICY == " DHT_DIR "\n"
         "        [dht_alpha <search factor> (default: %u)]\n"
         "        [dht_k <replication factor> (default: %u)]\n"
@@ -152,7 +153,7 @@ static void usage(void)
                /* unicast */
                FLAT_RANDOM, DHT_DIR, SHA3_256, LINK_STATE, MB_ECN_CA,
                /* dt */
-               DT(addr_size), DT(eid_size), DT(max_ttl),
+               DT(addr_size), DT(eid_size), DT(max_ttl), DT(max_rtt),
                /* dht */
                DHT(alpha), DHT(k), DHT(t_expire),
                DHT(t_refresh), DHT(t_replicate),
@@ -184,6 +185,7 @@ int do_bootstrap_ipcp(int     argc,
         uint8_t                 addr_size      = DT(addr_size);
         uint8_t                 eid_size       = DT(eid_size);
         uint8_t                 max_ttl        = DT(max_ttl);
+        uint16_t                max_rtt        = DT(max_rtt);
         struct routing_config   routing        = default_routing_config;
         enum pol_addr_auth      addr_auth_type = UNI(addr_auth_type);
         enum pol_cong_avoid     cong_avoid     = UNI(cong_avoid);
@@ -250,6 +252,8 @@ int do_bootstrap_ipcp(int     argc,
                         eid_size = atoi(*(argv + 1));
                 } else if (matches(*argv, "ttl") == 0) {
                         max_ttl = atoi(*(argv + 1));
+                } else if (matches(*argv, "rtt") == 0) {
+                        max_rtt = atoi(*(argv + 1));
                 } else if (matches(*argv, "port") == 0) {
                         port4 = atoi(*(argv + 1));
                         port6 = port4;
@@ -422,6 +426,7 @@ int do_bootstrap_ipcp(int     argc,
                                 conf.unicast.dt.addr_size    = addr_size;
                                 conf.unicast.dt.eid_size     = eid_size;
                                 conf.unicast.dt.max_ttl      = max_ttl;
+                                conf.unicast.dt.max_rtt      = max_rtt;
                                 conf.unicast.dt.routing      = routing;
                                 conf.unicast.addr_auth_type  = addr_auth_type;
                                 conf.unicast.cong_avoid      = cong_avoid;