ipcpd: Use capacity queue estimation for mb-ecn

The mb-ecn algorithm was using rbuff queue depths in packets to mark,
but sockets in the poa component report capacity in bytes. The tx
rings are now adaptive to block on queuing delay instead of when full
to prevent buffer bloat, controllable via fccntl (FLOWSTXQDLY and
FLOWGTXQDLY).

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

10 files changed, 1411 insertions, 43 deletions

diff --git a/src/lib/cap.c b/src/lib/cap.c
new file mode 100644
index 00000000..f116bfb0
--- /dev/null
+++ b/src/lib/cap.c
@@ -0,0 +1,187 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Link capacity estimation
+ *
+ *    Dimitri Staessens <dimitri@ouroboros.rocks>
+ *    Sander Vrijders   <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+/*
+ * Link-capacity estimation by watching the egress queue drain.
+ *
+ * A saturated link drains its queue at exactly its capacity, so we
+ * estimate capacity by measuring the drain rate of the transmit
+ * queue toward an n-1 flow (the flow to the layer below) while that
+ * queue is backlogged.
+ *
+ * Sampling is lock-free and off the fast path: the queue depth is
+ * read only at enqueue time, concurrently by many sender threads.
+ * Each enqueue bumps relaxed counters (packets, bytes, empty-queue
+ * hits). At most once per CAP_T_MIN, one thread wins a try-lock and
+ * closes a measurement window.
+ *
+ * Over a window, byte conservation gives the bytes that drained:
+ *   drained = queue at start (q0) + enqueued - queue now (q1)
+ * A window stays open until CAP_N_MIN packets' worth has drained, so
+ * its length self-scales with the link rate (~1 ms at 1 Gbit, ~19 ms
+ * at 10 Mbit). CAP_T_MAX discards a window that spanned a traffic gap.
+ *
+ * Only a backlogged link measures its own capacity, so a window
+ * whose ring ran mostly idle is discarded (a few empty samples, as
+ * from a token-bucket shaper, are tolerated). The drain rate feeds a
+ * max filter that jumps up at once but decays slowly, converging on
+ * the capacity from below. A window that touched an empty ring at
+ * either edge may have drained into downstream buffers faster than
+ * the wire, so it may only lower the estimate, never raise it.
+ */
+
+#if defined(__linux__) || defined(__CYGWIN__)
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+#else
+#ifndef _POSIX_C_SOURCE
+#define _POSIX_C_SOURCE 200809L
+#endif
+#endif
+
+#include "config.h"
+
+#include <ouroboros/atomics.h>
+#include <ouroboros/time.h>
+
+#include "cap.h"
+
+#include <string.h>
+
+#define CAP_T_MIN    (BILLION / 1000)     /* min close spacing ~1 ms  */
+#define CAP_T_MAX    (1ULL << 27)         /* voiding traffic gap ~134 ms */
+#define CAP_N_MIN    16                   /* drained packets to close */
+#define CAP_DEC_SHFT 4                    /* max-filter decay 1/16    */
+#define CAP_IDL_SHFT 3                    /* idle tolerance 1/8       */
+
+/* Busy-flag try-lock: test-and-set acquire, store release.           */
+#define CAP_TAS(p) __atomic_exchange_n(p, 1, __ATOMIC_ACQUIRE)
+#define CAP_REL(p) (__atomic_store_n(p, 0, __ATOMIC_RELEASE))
+
+void cap_clear(struct cap_est * e)
+{
+        memset(e, 0, sizeof(*e));
+}
+
+uint64_t cap_rate(const struct cap_est * e)
+{
+        return LOAD_RELAXED(&e->est);
+}
+
+/* Busy flag held; q1 is the caller's pre-write ring sample. */
+static void cap_close(struct cap_est * e,
+                      uint64_t         q1,
+                      uint64_t         now,
+                      uint64_t         gap)
+{
+        uint64_t pkt;              /* current c_pkt snapshot      */
+        uint64_t byt;              /* current c_byt snapshot      */
+        uint64_t idl;              /* current c_idl snapshot      */
+        uint64_t dt;               /* window duration (ns)        */
+        uint64_t enq;              /* packets enqueued in window  */
+        uint64_t avg;              /* mean packet size (bytes)    */
+        uint64_t r;                /* window drain rate (bytes/s) */
+        int64_t  drained;          /* bytes drained over window   */
+
+        pkt = LOAD_RELAXED(&e->c_pkt);
+        byt = LOAD_RELAXED(&e->c_byt);
+        idl = LOAD_RELAXED(&e->c_idl);
+
+        dt  = now - e->t0;
+        enq = pkt - e->pkt0;
+
+        drained = (int64_t) (e->q0 + (byt - e->byt0) - q1);
+
+        if (e->t0 == 0 || enq == 0)
+                goto reopen;
+
+        if (gap > CAP_T_MAX)
+                goto reopen; /* traffic stopped: window void */
+
+        avg = (byt - e->byt0) / enq;
+        if (drained < (int64_t) (CAP_N_MIN * avg))
+                return; /* extend the window until enough drains */
+
+        if ((idl - e->idl0) << CAP_IDL_SHFT > enq)
+                goto reopen; /* mostly idle ring: not saturated */
+
+        r = (uint64_t) drained * MILLION / (dt / 1000);
+        if (r >= e->rate) {
+                if (e->q0 > 0 && q1 > 0) /* empty edge drains below */
+                        e->rate = r;
+        } else {
+                e->rate -= (e->rate - r) >> CAP_DEC_SHFT;
+        }
+
+        STORE_RELAXED(&e->est, e->rate);
+ reopen:
+        e->t0   = now;
+        e->q0   = q1;
+        e->pkt0 = pkt;
+        e->byt0 = byt;
+        e->idl0 = idl;
+}
+
+void cap_update_at(struct cap_est * e,
+                   size_t           qlen,
+                   size_t           len,
+                   uint64_t         now)
+{
+        uint64_t prev;
+
+        FETCH_ADD_RELAXED(&e->c_pkt, 1);
+        FETCH_ADD_RELAXED(&e->c_byt, len);
+
+        if (qlen == 0)
+                FETCH_ADD_RELAXED(&e->c_idl, 1);
+
+        prev = LOAD_RELAXED(&e->t_last);
+        if (prev > now)
+                prev = now; /* a racing writer stamped ahead */
+
+        STORE_RELAXED(&e->t_last, now);
+
+        if (now - LOAD_RELAXED(&e->t_gate) < CAP_T_MIN)
+                return;
+
+        if (CAP_TAS(&e->busy) != 0)
+                return;
+
+        if (now - e->t_gate >= CAP_T_MIN) {
+                cap_close(e, qlen, now, now - prev);
+                STORE_RELAXED(&e->t_gate, now);
+        }
+
+        CAP_REL(&e->busy);
+}
+
+void cap_update(struct cap_est * e,
+                size_t           qlen,
+                size_t           len)
+{
+        struct timespec now;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+        cap_update_at(e, qlen, len, TS_TO_UINT64(now));
+}
diff --git a/src/lib/cap.h b/src/lib/cap.h
new file mode 100644
index 00000000..3d94d9a3
--- /dev/null
+++ b/src/lib/cap.h
@@ -0,0 +1,63 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Link capacity estimation
+ *
+ *    Dimitri Staessens <dimitri@ouroboros.rocks>
+ *    Sander Vrijders   <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#ifndef OUROBOROS_LIB_CAP_H
+#define OUROBOROS_LIB_CAP_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+#define CAP_ALIGN 64
+
+struct cap_est {
+        uint64_t c_pkt;  /* total packets enqueued (relaxed)  */
+        uint64_t c_byt;  /* total bytes enqueued (relaxed)    */
+        uint64_t c_idl;  /* times ring seen empty (relaxed)   */
+
+        uint64_t t_gate; /* last window close (ns)            */
+        uint64_t t_last; /* last update, to spot a gap (ns)   */
+        uint8_t  busy;   /* close in progress (try-lock)      */
+
+        uint64_t t0;     /* window start (ns), 0 = no window  */
+        uint64_t q0;     /* ring occupancy at window start    */
+        uint64_t pkt0;   /* c_pkt snapshot at window start    */
+        uint64_t byt0;   /* c_byt snapshot at window start    */
+        uint64_t idl0;   /* c_idl snapshot at window start    */
+        uint64_t rate;   /* filtered drain rate (bytes/s)     */
+
+        uint64_t est;    /* published estimate (bytes/s)      */
+} __attribute__((aligned(CAP_ALIGN)));
+
+void     cap_clear(struct cap_est * e);
+
+void     cap_update(struct cap_est * e,
+                    size_t           qlen,
+                    size_t           len);
+
+void     cap_update_at(struct cap_est * e,
+                       size_t           qlen,
+                       size_t           len,
+                       uint64_t         now);
+
+uint64_t cap_rate(const struct cap_est * e);
+
+#endif /* OUROBOROS_LIB_CAP_H */
diff --git a/src/lib/config.h.in b/src/lib/config.h.in
index 3c6985c3..62925a48 100644
--- a/src/lib/config.h.in
+++ b/src/lib/config.h.in
@@ -48,6 +48,7 @@
 #cmakedefine                     HAVE_PMULL
 
 #define SHM_LOCKFILE_NAME        "@SHM_LOCKFILE_NAME@"
+#define SSM_RBUFF_TXQ_DELAY       @SSM_RBUFF_TXQ_DELAY@ /* ms */
 #define FLOW_ALLOC_TIMEOUT        @FLOW_ALLOC_TIMEOUT@
 
 #define TPM_DEBUG_REPORT_INTERVAL @TPM_DEBUG_REPORT_INTERVAL@
diff --git a/src/lib/dev.c b/src/lib/dev.c
index eb706691..bce64c37 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -27,6 +27,7 @@
 #endif
 
 #include "config.h"
+#include "cap.h"
 #include "ssm.h"
 #include "poa/poa.h"
 
@@ -60,6 +61,7 @@
 #include <ouroboros/ssm_flow_set.h>
 #include <ouroboros/ssm_pool.h>
 #include <ouroboros/ssm_rbuff.h>
+#include <ouroboros/time.h>
 #include <ouroboros/tw.h>
 #include <ouroboros/utils.h>
 
@@ -85,6 +87,7 @@
 #define DONE_PART -2
 
 #define CRCLEN    (sizeof(uint32_t))
+#define FLOW_AVG_SHIFT 3
 #define SECMEMSZ  16384
 #define MSGBUFSZ  2048
 
@@ -123,7 +126,13 @@ struct flow {
 
         struct frcti *        frcti;
 
+        /* Mean written packet size (bytes), EWMA over the send path. */
+        size_t                mean_len;
+
         struct poa_flow *     poa;   /* NULL for shared memory flows */
+
+        /* Egress capacity estimator; armed by the IPCP, else NULL. */
+        struct cap_est *      cap;
 };
 
 struct flow_set {
@@ -730,6 +739,8 @@ static void do_flow_fini(int fd)
 
         crypt_destroy_ctx(proc.flows[fd].crypt);
 
+        free(proc.flows[fd].cap);
+
         flow_clear(fd);
 }
 
@@ -768,6 +779,7 @@ static int flow_init(struct flow_info * info,
                      struct poa_flow *  pf)
 {
         struct timespec now;
+        struct timespec txq;
         struct flow *   flow;
         int             fd;
         int             err = -ENOMEM;
@@ -795,6 +807,11 @@ static int flow_init(struct flow_info * info,
                 if (flow->tx_rb == NULL)
                         goto fail_tx_rb;
 
+                txq.tv_sec  = SSM_RBUFF_TXQ_DELAY / 1000;
+                txq.tv_nsec = (SSM_RBUFF_TXQ_DELAY % 1000) * MILLION;
+
+                ssm_rbuff_set_txq_target(flow->tx_rb, &txq);
+
                 flow->set = ssm_flow_set_open(info->n_1_pid);
                 if (flow->set == NULL)
                         goto fail_set;
@@ -1475,6 +1492,25 @@ int fccntl(int fd,
                         goto einval;
                 *maxp = flow_user_mtu(flow, flow->info.mtu);
                 break;
+        case FLOWSTXQDLY:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+
+                if (flow->tx_rb == NULL)
+                        goto eperm;
+
+                ssm_rbuff_set_txq_target(flow->tx_rb, timeo);
+                break;
+        case FLOWGTXQDLY:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+
+                if (flow->tx_rb == NULL)
+                        goto eperm;
+                ssm_rbuff_get_txq_target(flow->tx_rb, timeo);
+                break;
         case FLOWSFLAGS:
                 old_acc = flow->oflags & FLOWFACCMODE;
                 flow->oflags = va_arg(l, uint32_t);
@@ -1609,6 +1645,25 @@ int fccntl(int fd,
         return -EPERM;
 }
 
+/*
+ * The ring counts slots, so the queue is only bytes if we know what a
+ * packet weighs. Ordered so the unsigned arithmetic cannot wrap.
+ */
+static void flow_mean_len_update(struct flow * flow,
+                                 size_t        len)
+{
+        size_t avg = LOAD_RELAXED(&flow->mean_len);
+
+        if (avg == 0) {
+                STORE_RELAXED(&flow->mean_len, len);
+                return;
+        }
+
+        avg = avg + (len >> FLOW_AVG_SHIFT) - (avg >> FLOW_AVG_SHIFT);
+
+        STORE_RELAXED(&flow->mean_len, avg == 0 ? 1 : avg);
+}
+
 static int flow_tx_spb(struct flow *        flow,
                        struct ssm_pk_buff * spb,
                        uint16_t             flags,
@@ -1643,6 +1698,8 @@ static int flow_tx_spb(struct flow *        flow,
         if (flow->poa != NULL)
                 return poa_flow_tx(flow->poa, spb, block, abstime);
 
+        flow_mean_len_update(flow, ssm_pk_buff_len(spb));
+
         if (!block)
                 ret = ssm_rbuff_write(flow->tx_rb, idx);
         else
@@ -2941,25 +2998,92 @@ size_t ipcp_flow_queued(int fd)
         if (proc.flows[fd].poa != NULL)
                 return poa_flow_qlen(proc.flows[fd].poa);
 
-        return ssm_rbuff_queued(proc.flows[fd].tx_rb);
+        return ssm_rbuff_queued(proc.flows[fd].tx_rb)
+                * LOAD_RELAXED(&proc.flows[fd].mean_len);
 }
 
-int ipcp_flow_queue_id(int fd)
+size_t ipcp_flow_mean_len(int fd)
 {
-        int qid;
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        if (proc.flows[fd].poa != NULL)
+                return poa_flow_mean_len(proc.flows[fd].poa);
+
+        return LOAD_RELAXED(&proc.flows[fd].mean_len);
+}
+
+/* An update racing the arm seeds one bogus window; the filter absorbs. */
+int ipcp_flow_cap_arm(int fd)
+{
+        struct flow *    flow;
+        struct cap_est * e;
+
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL) {
+                cap_clear(poa_flow_cap_est(flow->poa));
+                return 0;
+        }
+
+        e = flow->cap;
+        if (e != NULL) {
+                cap_clear(e);
+                return 0;
+        }
+
+        if (posix_memalign((void **) &e, CAP_ALIGN, sizeof(*e)) != 0)
+                return -ENOMEM;
+
+        cap_clear(e);
+
+        STORE_RELEASE(&flow->cap, e);
+
+        return 0;
+}
+
+void ipcp_flow_cap_update(int    fd,
+                          size_t qlen,
+                          size_t len)
+{
+        struct flow *    flow;
+        struct cap_est * e;
 
         assert(fd >= 0 && fd < PROC_MAX_FLOWS);
         assert(proc.flows[fd].info.id >= 0);
 
-        if (proc.flows[fd].poa == NULL)
-                return fd;
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL) {
+                cap_update(poa_flow_cap_est(flow->poa), qlen, len);
+                return;
+        }
+
+        e = LOAD_ACQUIRE(&flow->cap);
+        if (e == NULL)
+                return;
 
-        /* An unidentified PoA answers for itself, never for an fd. */
-        qid = poa_flow_qid(proc.flows[fd].poa);
-        if (qid < 0 || qid >= POA_MAX_POAS)
-                return fd;
+        cap_update(e, qlen, len);
+}
+
+uint64_t ipcp_flow_cap(int fd)
+{
+        struct flow *    flow;
+        struct cap_est * e;
+
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL)
+                return cap_rate(poa_flow_cap_est(flow->poa));
+
+        e = LOAD_ACQUIRE(&flow->cap);
+        if (e == NULL)
+                return 0;
 
-        return PROC_MAX_FLOWS + qid;
+        return cap_rate(e);
 }
 
 int local_flow_transfer(int               src_fd,
@@ -3028,4 +3152,5 @@ int local_flow_transfer(int               src_fd,
         return ret;
 }
 
+#include "cap.c"
 #include "poa/poa.c"
diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c
index 3ad17c4f..b40d9fea 100644
--- a/src/lib/poa/poa.c
+++ b/src/lib/poa/poa.c
@@ -1678,9 +1678,9 @@ size_t poa_flow_mean_len(const struct poa_flow * pf)
         return LOAD_RELAXED(&pf->poa->avg_len);
 }
 
-int poa_flow_qid(const struct poa_flow * pf)
+struct cap_est * poa_flow_cap_est(struct poa_flow * pf)
 {
-        return pf->poa->qid;
+        return &pf->poa->cap;
 }
 
 void poa_flow_ready(struct poa_flow * pf)
@@ -1830,7 +1830,6 @@ static struct poa * poa_create(enum poa_type          type,
         poa->ops    = ops;
         poa->mpl    = ops->mpl;
         poa->n_eids = n_eids;
-        poa->qid    = -1;
 
         return poa;
 
@@ -2213,30 +2212,15 @@ void poa_fini(void)
         pthread_mutex_destroy(&poas.mtx);
 }
 
-/*
- * Lowest queue id no attached PoA holds; detaching frees it by
- * leaving the list. Caller holds poas.lock.
- */
-static int poa_qid_alloc(void)
+static size_t poa_count(void)
 {
         struct list_head * p;
-        bool               used[POA_MAX_POAS];
-        int                i;
-
-        memset(used, 0, sizeof(used));
-
-        list_for_each(p, &poas.list) {
-                struct poa * poa = list_entry(p, struct poa, next);
-
-                if (poa->qid >= 0 && poa->qid < POA_MAX_POAS)
-                        used[poa->qid] = true;
-        }
+        size_t             n = 0;
 
-        for (i = 0; i < POA_MAX_POAS; i++)
-                if (!used[i])
-                        return i;
+        list_for_each(p, &poas.list)
+                n++;
 
-        return -1;
+        return n;
 }
 
 static int poa_add(const struct poa_spec * spec,
@@ -2275,8 +2259,7 @@ static int poa_add(const struct poa_spec * spec,
 
         pthread_rwlock_wrlock(&poas.lock);
 
-        poa->qid = poa_qid_alloc();
-        if (poa->qid < 0) {
+        if (poa_count() >= POA_MAX_POAS) {
                 pthread_rwlock_unlock(&poas.lock);
                 goto fail_start;
         }
diff --git a/src/lib/poa/poa.h b/src/lib/poa/poa.h
index 014986a3..9edb0335 100644
--- a/src/lib/poa/poa.h
+++ b/src/lib/poa/poa.h
@@ -34,6 +34,8 @@
 #include <ouroboros/time.h>
 #include <ouroboros/utils.h>
 
+#include "../cap.h"
+
 #include <errno.h>
 #include <limits.h>
 #include <poll.h>
@@ -191,9 +193,6 @@ struct poa {
 
         time_t                 mpl;
 
-        /* Identifies the transmit queue the flows on this PoA share. */
-        int                    qid;
-
         /* Mean sent packet size (bytes), EWMA over the send path. */
         size_t                 avg_len;
         /* Cost of one packet in the queue, in the transport's terms. */
@@ -204,6 +203,9 @@ struct poa {
         size_t                 q_cache;
         uint64_t               q_time;
 
+        /* Capacity estimator of the queue the flows on this PoA share. */
+        struct cap_est         cap;
+
         /* Queued management frames, capped; poas.mgmt_mtx guards. */
         size_t                 n_mgmt;
 
@@ -245,7 +247,7 @@ size_t            poa_flow_qlen(const struct poa_flow * pf);
 
 size_t            poa_flow_qpkts(const struct poa_flow * pf);
 
-int               poa_flow_qid(const struct poa_flow * pf);
+struct cap_est *  poa_flow_cap_est(struct poa_flow * pf);
 
 size_t            poa_flow_mean_len(const struct poa_flow * pf);
 
diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c
index 04978d82..e35a27a9 100644
--- a/src/lib/ssm/rbuff.c
+++ b/src/lib/ssm/rbuff.c
@@ -57,6 +57,8 @@
 #define LOAD_ACQUIRE(ptr) (__atomic_load_n(ptr, __ATOMIC_ACQUIRE))
 #define STORE_RELEASE(ptr, val)                                                \
         (__atomic_store_n(ptr, val, __ATOMIC_RELEASE))
+#define STORE_RELAXED(ptr, val)                                                \
+        (__atomic_store_n(ptr, val, __ATOMIC_RELAXED))
 
 #define HEAD(rb)       (rb->shm_base[LOAD_RELAXED(rb->head)])
 #define TAIL(rb)       (rb->shm_base[LOAD_RELAXED(rb->tail)])
@@ -70,6 +72,20 @@
 #define IS_FULL(rb)    (QUEUED(rb) == (SSM_RBUFF_SIZE - 1))
 #define IS_EMPTY(rb)   (HEAD_IDX(rb) == TAIL_IDX(rb))
 
+/*
+ * Occupancy limiter: bound a tx ring by queueing delay instead of
+ * slot count, so a slow link does not accumulate seconds of backlog.
+ * A zero target is unlimited: the wait predicate then reduces to
+ * physical fullness. A ring is unlimited until a target is set.
+ */
+#define TXQ_MIN_SLOTS   4              /* floor: jitter margin       */
+#define TXQ_PRIO_MUL    2              /* headroom kept for retx      */
+#define TXQ_SHIFT       2              /* EWMA weight 1/4            */
+#define TXQ_SAMPLE_MASK 15             /* resample every 16 writes   */
+#define TXQ_MIN_DT_NS   10000LL        /* skip sub-10us samples      */
+
+#define TXQ_UNLIMITED (SSM_RBUFF_SIZE - 1)
+
 struct ssm_rbuff {
         ssize_t *         shm_base;     /* start of shared memory   */
         size_t *          head;         /* start of ringbuffer      */
@@ -81,8 +97,20 @@ struct ssm_rbuff {
         pid_t             pid;          /* pid of the owner         */
         int               flow_id;      /* flow_id of the flow      */
         size_t            n_users;      /* in-flight users          */
+        uint64_t          txq_target;   /* target queue delay, ns   */
+        size_t            txq_limit;    /* current occupancy limit  */
+        int64_t           txq_rate;     /* EWMA drain rate, slots/s */
+        uint64_t          txq_ns;       /* last sample time, ns     */
+        size_t            txq_wr;       /* writes since last sample */
+        size_t            txq_q0;       /* queued count at sample   */
 };
 
+#define TXQ_ON(rb)    (LOAD_RELAXED(&(rb)->txq_target) != 0)
+#define TXQ_LIMIT(rb) (TXQ_ON(rb) ? LOAD_RELAXED(&(rb)->txq_limit)         \
+                                  : TXQ_UNLIMITED)
+#define OVER_LIMIT(rb) (QUEUED(rb) >= TXQ_LIMIT(rb))
+
+
 #define MM_FLAGS (PROT_READ | PROT_WRITE)
 
 static struct ssm_rbuff * rbuff_create(pid_t pid,
@@ -121,6 +149,12 @@ static struct ssm_rbuff * rbuff_create(pid_t pid,
         rb->pid      = pid;
         rb->flow_id  = flow_id;
         rb->n_users  = 0;
+        rb->txq_target = 0;         /* unlimited until set */
+        rb->txq_limit = TXQ_UNLIMITED;
+        rb->txq_rate = 0;
+        rb->txq_ns   = 0;
+        rb->txq_wr   = 0;
+        rb->txq_q0   = 0;
 
         return rb;
 
@@ -251,8 +285,106 @@ static void __cleanup_rbuff_reader(void * o)
         __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
 }
 
-int ssm_rbuff_write(struct ssm_rbuff * rb,
-                    size_t             off)
+/*
+ * Refresh the drain-rate estimate and derived occupancy limit.
+ * Called with rb->mtx held, at most once per TXQ_SAMPLE_MASK writes.
+ */
+static void rbuff_txq_sample(struct ssm_rbuff * rb,
+                             size_t             queued)
+{
+        struct timespec now;
+        uint64_t        now_ns;
+        uint64_t        last_ns;
+        int64_t         dt_ns;
+        int64_t         written;
+        int64_t         grown;
+        int64_t         drained;
+        int64_t         sample_rate;
+        int64_t         rate;
+        int64_t         target;
+        size_t          limit;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+        now_ns = TS_TO_UINT64(now);
+
+        last_ns = LOAD_RELAXED(&rb->txq_ns);
+        if (last_ns == 0) {
+                /* No prior sample: seed and stay at the default limit. */
+                STORE_RELAXED(&rb->txq_ns, now_ns);
+                STORE_RELAXED(&rb->txq_q0, queued);
+                STORE_RELAXED(&rb->txq_wr, 0);
+                return;
+        }
+
+        dt_ns = (int64_t) (now_ns - last_ns);
+        if (dt_ns < TXQ_MIN_DT_NS)
+                return;
+
+        written = (int64_t) LOAD_RELAXED(&rb->txq_wr);
+        grown   = (int64_t) queued - (int64_t) LOAD_RELAXED(&rb->txq_q0);
+
+        drained = written - grown;
+        if (drained < 0)
+                drained = 0;
+
+        sample_rate = drained * BILLION / dt_ns;
+
+        rate  = LOAD_RELAXED(&rb->txq_rate);
+        rate += (sample_rate - rate) >> TXQ_SHIFT;
+        if (rate < 0)
+                rate = 0;
+
+        target = (int64_t) LOAD_RELAXED(&rb->txq_target);
+
+        limit  = (size_t) (rate * target / BILLION);
+        if (limit < TXQ_MIN_SLOTS)
+                limit = TXQ_MIN_SLOTS;
+
+        if (limit > TXQ_UNLIMITED)
+                limit = TXQ_UNLIMITED;
+
+        STORE_RELAXED(&rb->txq_rate, rate);
+        STORE_RELAXED(&rb->txq_limit, limit);
+        STORE_RELAXED(&rb->txq_ns, now_ns);
+        STORE_RELAXED(&rb->txq_q0, queued);
+        STORE_RELAXED(&rb->txq_wr, 0);
+}
+
+/* Bumps the write counter, resampling every TXQ_SAMPLE_MASK writes. */
+static void rbuff_txq_touch(struct ssm_rbuff * rb)
+{
+        size_t wr;
+
+        wr = LOAD_RELAXED(&rb->txq_wr) + 1;
+
+        STORE_RELAXED(&rb->txq_wr, wr);
+
+        if ((wr & TXQ_SAMPLE_MASK) == 0)
+                rbuff_txq_sample(rb, QUEUED(rb));
+}
+
+/*
+ * A retransmission outranks new data but stays bounded: its ceiling is
+ * a multiple of the limit, so the headroom above it is reserved and the
+ * queueing delay stays within a known factor of the target.
+ */
+static size_t rbuff_txq_prio_limit(struct ssm_rbuff * rb)
+{
+        size_t lim;
+
+        if (!TXQ_ON(rb))
+                return TXQ_UNLIMITED;
+
+        lim = LOAD_RELAXED(&rb->txq_limit) * TXQ_PRIO_MUL;
+
+        return lim > TXQ_UNLIMITED ? TXQ_UNLIMITED : lim;
+}
+
+/* prio outranks new data up to its own, higher, ceiling. */
+static int rbuff_write_nb(struct ssm_rbuff * rb,
+                          size_t             off,
+                          bool               prio)
 {
         size_t flags;
         bool   was_empty;
@@ -276,7 +408,8 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
 
         robust_mutex_lock(rb->mtx);
 
-        if (IS_FULL(rb)) {
+        if (QUEUED(rb) >= (prio ? rbuff_txq_prio_limit(rb)
+                                : TXQ_LIMIT(rb))) {
                 ret = -EAGAIN;
                 goto fail_mutex;
         }
@@ -289,6 +422,10 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
         if (was_empty)
                 pthread_cond_broadcast(rb->add);
 
+        /* Only an enqueue feeds the estimator; a refusal wrote nothing. */
+        if (TXQ_ON(rb))
+                rbuff_txq_touch(rb);
+
         pthread_mutex_unlock(rb->mtx);
 
         __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
@@ -301,6 +438,19 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
         return ret;
 }
 
+int ssm_rbuff_write(struct ssm_rbuff * rb,
+                    size_t             off)
+{
+        return rbuff_write_nb(rb, off, false);
+}
+
+/* For a packet the peer is already waiting on; skips the limit. */
+int ssm_rbuff_write_prio(struct ssm_rbuff * rb,
+                         size_t             off)
+{
+        return rbuff_write_nb(rb, off, true);
+}
+
 int ssm_rbuff_write_b(struct ssm_rbuff *      rb,
                       size_t                  off,
                       const struct timespec * abstime)
@@ -329,7 +479,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff *      rb,
 
         pthread_cleanup_push(__cleanup_rbuff_reader, rb);
 
-        while (IS_FULL(rb) && ret != -ETIMEDOUT) {
+        while (OVER_LIMIT(rb) && ret != -ETIMEDOUT) {
                 flags = __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
                 if (flags & RB_FLOWDOWN) {
                         ret = -EFLOWDOWN;
@@ -346,6 +496,9 @@ int ssm_rbuff_write_b(struct ssm_rbuff *      rb,
                 ADVANCE_HEAD(rb);
                 if (was_empty)
                         pthread_cond_broadcast(rb->add);
+
+                if (TXQ_ON(rb))
+                        rbuff_txq_touch(rb);
         }
 
         pthread_mutex_unlock(rb->mtx);
@@ -484,6 +637,40 @@ uint32_t ssm_rbuff_get_flags(struct ssm_rbuff * rb)
         return (uint32_t) __atomic_load_n(rb->flags, __ATOMIC_SEQ_CST);
 }
 
+/* Current occupancy limit; SSM_RBUFF_SIZE - 1 when unlimited. */
+size_t ssm_rbuff_get_limit(struct ssm_rbuff * rb)
+{
+        assert(rb != NULL);
+
+        return TXQ_LIMIT(rb);
+}
+
+/* Target queueing delay; a zero target is unlimited. */
+void ssm_rbuff_set_txq_target(struct ssm_rbuff *      rb,
+                              const struct timespec * ts)
+{
+        assert(rb != NULL);
+        assert(ts != NULL);
+
+        STORE_RELAXED(&rb->txq_limit, TXQ_UNLIMITED);
+        STORE_RELAXED(&rb->txq_rate, 0);
+        STORE_RELAXED(&rb->txq_ns, 0);
+        STORE_RELAXED(&rb->txq_wr, 0);
+        STORE_RELAXED(&rb->txq_q0, 0);
+
+        STORE_RELAXED(&rb->txq_target, TS_TO_UINT64(*ts));
+}
+
+/* Current target queueing delay for the tx occupancy limiter. */
+void ssm_rbuff_get_txq_target(struct ssm_rbuff * rb,
+                              struct timespec *  ts)
+{
+        assert(rb != NULL);
+        assert(ts != NULL);
+
+        UINT64_TO_TS(LOAD_RELAXED(&rb->txq_target), ts);
+}
+
 void ssm_rbuff_fini(struct ssm_rbuff * rb)
 {
         assert(rb != NULL);
diff --git a/src/lib/ssm/tests/rbuff_test.c b/src/lib/ssm/tests/rbuff_test.c
index 48e5a714..57e6198e 100644
--- a/src/lib/ssm/tests/rbuff_test.c
+++ b/src/lib/ssm/tests/rbuff_test.c
@@ -34,6 +34,9 @@
 #include <ouroboros/errno.h>
 #include <ouroboros/time.h>
 
+/* Mirrors TXQ_MIN_SLOTS in ssm/rbuff.c; keep in sync. */
+#define FLOOR_SLOTS 4
+
 #include <errno.h>
 #include <stdio.h>
 #include <unistd.h>
@@ -652,6 +655,389 @@ static int test_ssm_rbuff_threaded(void)
         return TEST_RC_FAIL;
 }
 
+static int test_ssm_rbuff_limit_off(void)
+{
+        struct ssm_rbuff * rb;
+        size_t             i;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 11);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        if (ssm_rbuff_get_limit(rb) != SSM_RBUFF_SIZE - 1) {
+                printf("Expected default limit %d, got %zu.\n",
+                       SSM_RBUFF_SIZE - 1, ssm_rbuff_get_limit(rb));
+                goto fail_rb;
+        }
+
+        for (i = 0; i < SSM_RBUFF_SIZE - 1; ++i) {
+                if (ssm_rbuff_write(rb, i) < 0) {
+                        printf("Failed to write at index %zu.\n", i);
+                        goto fail_rb;
+                }
+        }
+
+        if (ssm_rbuff_write(rb, 999) != -EAGAIN) {
+                printf("Expected -EAGAIN on physically full buffer.\n");
+                goto fail_rb;
+        }
+
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_ssm_rbuff_limit_slow(void)
+{
+        struct ssm_rbuff * rb;
+        struct timespec    dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION};
+        struct timespec    delay = {0, 10 * MILLION};
+        size_t             limit;
+        size_t             i;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 12);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+
+        for (i = 0; i < 32; ++i) {
+                if (ssm_rbuff_write_b(rb, i, NULL) < 0) {
+                        printf("Failed to write at index %zu.\n", i);
+                        goto fail_rb;
+                }
+                nanosleep(&delay, NULL);
+
+                if (ssm_rbuff_read(rb) < 0) {
+                        printf("Failed to read at index %zu.\n", i);
+                        goto fail_rb;
+                }
+        }
+
+        limit = ssm_rbuff_get_limit(rb);
+        if (limit > FLOOR_SLOTS) {
+                printf("Expected limit near the floor, got %zu.\n", limit);
+                goto fail_rb;
+        }
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_ssm_rbuff_limit_fast(void)
+{
+        struct ssm_rbuff * rb;
+        struct timespec    dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION};
+        size_t             limit;
+        size_t             i;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 13);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+
+        for (i = 0; i < 200; ++i) {
+                if (ssm_rbuff_write_b(rb, i, NULL) < 0) {
+                        printf("Failed to write at index %zu.\n", i);
+                        goto fail_rb;
+                }
+
+                if (ssm_rbuff_read(rb) < 0) {
+                        printf("Failed to read at index %zu.\n", i);
+                        goto fail_rb;
+                }
+        }
+
+        limit = ssm_rbuff_get_limit(rb);
+        if (limit != SSM_RBUFF_SIZE - 1) {
+                printf("Expected limit %d, got %zu.\n",
+                       SSM_RBUFF_SIZE - 1, limit);
+                goto fail_rb;
+        }
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_ssm_rbuff_limit_floor(void)
+{
+        struct ssm_rbuff * rb;
+        struct timespec    dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION};
+        struct timespec    interval = {0, 50 * MILLION};
+        struct timespec    now;
+        struct timespec    abs_timeout;
+        size_t             limit;
+        int                ret = 0;
+        size_t             i;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 14);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+        ts_add(&now, &interval, &abs_timeout);
+
+        for (i = 0; i < SSM_RBUFF_SIZE; ++i) {
+                ret = ssm_rbuff_write_b(rb, i, &abs_timeout);
+                if (ret == -ETIMEDOUT)
+                        break;
+
+                if (ret < 0) {
+                        printf("Write failed at index %zu: %d.\n", i, ret);
+                        goto fail_rb;
+                }
+        }
+
+        if (ret != -ETIMEDOUT) {
+                printf("Expected the limiter to block the ring.\n");
+                goto fail_rb;
+        }
+
+        limit = ssm_rbuff_get_limit(rb);
+        if (limit > FLOOR_SLOTS) {
+                printf("Expected floor limit, got %zu.\n", limit);
+                goto fail_rb;
+        }
+
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_ssm_rbuff_txq_target(void)
+{
+        struct ssm_rbuff * rb;
+        struct timespec    dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION};
+        struct timespec    delay = {0, 5 * MILLION};
+        struct timespec    small = {0, 2 * MILLION};
+        struct timespec    big   = {0, 200 * MILLION};
+        struct timespec    def;
+        struct timespec    got;
+        size_t             limit_small;
+        size_t             limit_big;
+        size_t             i;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 15);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        /* A fresh ring is unlimited; rx rings must not inherit a bound. */
+        ssm_rbuff_get_txq_target(rb, &got);
+        if (got.tv_sec != 0 || got.tv_nsec != 0) {
+                printf("A new ring is not unlimited.\n");
+                goto fail_rb;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+        ssm_rbuff_get_txq_target(rb, &def);
+
+        ssm_rbuff_set_txq_target(rb, &small);
+
+        for (i = 0; i < 64; ++i) {
+                if (ssm_rbuff_write_b(rb, i, NULL) < 0) {
+                        printf("Failed to write at index %zu.\n", i);
+                        goto fail_rb;
+                }
+                nanosleep(&delay, NULL);
+
+                if (ssm_rbuff_read(rb) < 0) {
+                        printf("Failed to read at index %zu.\n", i);
+                        goto fail_rb;
+                }
+        }
+
+        limit_small = ssm_rbuff_get_limit(rb);
+
+        ssm_rbuff_set_txq_target(rb, &big);
+
+        for (i = 0; i < 64; ++i) {
+                if (ssm_rbuff_write_b(rb, i, NULL) < 0) {
+                        printf("Failed to write at index %zu.\n", i);
+                        goto fail_rb;
+                }
+                nanosleep(&delay, NULL);
+
+                if (ssm_rbuff_read(rb) < 0) {
+                        printf("Failed to read at index %zu.\n", i);
+                        goto fail_rb;
+                }
+        }
+
+        limit_big = ssm_rbuff_get_limit(rb);
+        if (limit_big <= limit_small) {
+                printf("Expected a larger target to grow the limit: "
+                       "%zu -> %zu.\n", limit_small, limit_big);
+                goto fail_rb;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+        ssm_rbuff_get_txq_target(rb, &got);
+
+        if (got.tv_sec != def.tv_sec || got.tv_nsec != def.tv_nsec) {
+                printf("NULL did not restore the default target.\n");
+                goto fail_rb;
+        }
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_ssm_rbuff_write_over_limit(void)
+{
+        struct ssm_rbuff * rb;
+        struct timespec    dfl = {0, SSM_RBUFF_TXQ_DELAY * MILLION};
+        struct timespec    age = {0, 20 * 1000};
+        size_t             count;
+        int                ret = 0;
+
+        TEST_START();
+
+        rb = ssm_rbuff_create(getpid(), 16);
+        if (rb == NULL) {
+                printf("Failed to create rbuff.\n");
+                goto fail;
+        }
+
+        ssm_rbuff_set_txq_target(rb, &dfl);
+
+        for (count = 0; count < SSM_RBUFF_SIZE; ++count) {
+                ret = ssm_rbuff_write(rb, count);
+                if (ret == -EAGAIN)
+                        break;
+
+                if (ret < 0) {
+                        printf("Write failed at index %zu: %d.\n", count, ret);
+                        goto fail_rb;
+                }
+
+                /* Age the seed sample past the estimator's dt floor. */
+                if (count == 16)
+                        nanosleep(&age, NULL);
+        }
+
+        if (ret != -EAGAIN) {
+                printf("Expected the limiter to reject a write.\n");
+                goto fail_rb;
+        }
+
+        if (count >= SSM_RBUFF_SIZE / 2) {
+                printf("Expected -EAGAIN well before a full ring, "
+                       "got %zu writes.\n", count);
+                goto fail_rb;
+        }
+
+        if (ssm_rbuff_queued(rb) != count) {
+                printf("Queued %zu does not match write count %zu.\n",
+                       ssm_rbuff_queued(rb), count);
+                goto fail_rb;
+        }
+
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+
+ fail_rb:
+        while (ssm_rbuff_read(rb) >= 0)
+                ;
+
+        ssm_rbuff_destroy(rb);
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
 int rbuff_test(int     argc,
                char ** argv)
 {
@@ -670,6 +1056,12 @@ int rbuff_test(int     argc,
         ret |= test_ssm_rbuff_blocking();
         ret |= test_ssm_rbuff_blocking_timeout();
         ret |= test_ssm_rbuff_blocking_flowdown();
+        ret |= test_ssm_rbuff_limit_off();
+        ret |= test_ssm_rbuff_limit_slow();
+        ret |= test_ssm_rbuff_limit_fast();
+        ret |= test_ssm_rbuff_limit_floor();
+        ret |= test_ssm_rbuff_txq_target();
+        ret |= test_ssm_rbuff_write_over_limit();
 
         return ret;
 }
diff --git a/src/lib/tests/CMakeLists.txt b/src/lib/tests/CMakeLists.txt
index 1f2e9ba2..d470d539 100644
--- a/src/lib/tests/CMakeLists.txt
+++ b/src/lib/tests/CMakeLists.txt
@@ -10,6 +10,7 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c
   auth_test_slh_dsa.c
   bitmap_test.c
   btree_test.c
+  cap_test.c
   crypt_test.c
   poa_test.c
   hash_test.c
diff --git a/src/lib/tests/cap_test.c b/src/lib/tests/cap_test.c
new file mode 100644
index 00000000..ea0e1fef
--- /dev/null
+++ b/src/lib/tests/cap_test.c
@@ -0,0 +1,427 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Unit tests for link capacity estimation
+ *
+ *    Dimitri Staessens <dimitri@ouroboros.rocks>
+ *    Sander Vrijders   <sander@ouroboros.rocks>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#include "../cap.c"
+
+#include <test/test.h>
+
+#include <inttypes.h>
+#include <stdbool.h>
+
+#define TICK     (50 * 1000ULL) /* 50 us between packets     */
+#define LEN      1000ULL        /* default packet size (B)   */
+#define QLEN     (8 * LEN)      /* steady backlog (bytes)    */
+#define RATE     (LEN * BILLION / TICK) /* LEN per TICK = 20 MB/s */
+
+#define SHP_LEN  1250ULL        /* shaped-link packet (B)    */
+#define SHP_STEP 20             /* packets per shaped window */
+#define SHP_RATE (SHP_LEN * BILLION / (SHP_STEP * TICK))
+
+/* Draining CAP_N_MIN of these outlasts CAP_T_MAX without a gap. */
+#define LOW_STEP (250 * TICK)   /* 12.5 ms between packets   */
+#define LOW_RATE (LEN * BILLION / LOW_STEP)
+
+/* Within the quarter-log2 band the wire code publishes. */
+static bool rate_is_near(uint64_t got,
+                         uint64_t exp)
+{
+        return got >= exp - exp / 8 && got <= exp + exp / 8;
+}
+
+static int test_cap_est_clear(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        if (cap_rate(&e) != 0) {
+                printf("Fresh estimator not unknown.\n");
+                goto fail;
+        }
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, QLEN, LEN, i * TICK);
+
+        if (cap_rate(&e) == 0) {
+                printf("No estimate to clear.\n");
+                goto fail;
+        }
+
+        cap_clear(&e);
+
+        if (cap_rate(&e) != 0) {
+                printf("Clear did not drop the estimate.\n");
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/* 1000 B every 50 us, ring steady at 8: drain = 20 MB/s. */
+static int test_cap_est_busy_window(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, QLEN, LEN, i * TICK);
+
+        if (!rate_is_near(cap_rate(&e), RATE)) {
+                printf("Estimated rate: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_cap_est_idle_tolerated(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, i == 21 ? 0 : QLEN, LEN, i * TICK);
+
+        if (!rate_is_near(cap_rate(&e), RATE)) {
+                printf("Grazed window: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_cap_est_mostly_idle_rejects(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 100; i++)
+                cap_update_at(&e, 0, LEN, i * TICK);
+
+        if (cap_rate(&e) != 0) {
+                printf("Idle ring estimated %" PRIu64 ".\n", cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/* 1000 B every 100 us: 10 slots/ms closes on a 2 ms window. */
+static int test_cap_est_slow_link_extends(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 30; i++)
+                cap_update_at(&e, QLEN, LEN, i * 2 * TICK);
+
+        if (!rate_is_near(cap_rate(&e), RATE / 2)) {
+                printf("Slow link: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) (RATE / 2), cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/* 1250 B every ms; one empty observation per 20 packets. */
+static int test_cap_est_shaped_link(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 100; i++)
+                cap_update_at(&e, i % SHP_STEP == 0 ? 0 : 6 * SHP_LEN,
+                                  SHP_LEN, i * SHP_STEP * TICK);
+
+        if (!rate_is_near(cap_rate(&e), SHP_RATE)) {
+                printf("Shaped link: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) SHP_RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/* Open a window, trickle 4 slots, then ~200 ms of silence. */
+static int test_cap_est_stale_discard(void)
+{
+        struct cap_est e;
+        uint64_t       t;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 5; i++)
+                cap_update_at(&e, QLEN, LEN, i * CAP_T_MIN);
+
+        t = 205 * CAP_T_MIN;
+
+        cap_update_at(&e, QLEN, LEN, t);
+
+        if (cap_rate(&e) != 0) {
+                printf("Gap window estimated %" PRIu64 ".\n", cap_rate(&e));
+                goto fail;
+        }
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, QLEN, LEN, t + i * TICK);
+
+        if (!rate_is_near(cap_rate(&e), RATE)) {
+                printf("Post-gap: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_cap_est_empty_start_no_raise(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        cap_update_at(&e, 0, LEN, CAP_T_MIN);
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, QLEN, LEN, CAP_T_MIN + i * TICK);
+
+        if (cap_rate(&e) != 0) {
+                printf("Empty-start window raised to %" PRIu64 ".\n",
+                       cap_rate(&e));
+                goto fail;
+        }
+
+        for (i = 41; i <= 60; i++)
+                cap_update_at(&e, QLEN, LEN, CAP_T_MIN + i * TICK);
+
+        if (!rate_is_near(cap_rate(&e), RATE)) {
+                printf("Backlogged window: exp %" PRIu64 ", got %" PRIu64
+                       ".\n", (uint64_t) RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/*
+ * Max filter: fast attack on a high sample, slow release on the
+ * lower samples from a halved packet size (10 MB/s).
+ */
+static int test_cap_est_max_filter(void)
+{
+        struct cap_est e;
+        uint64_t       high;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 40; i++)
+                cap_update_at(&e, QLEN, LEN, i * TICK);
+
+        high = cap_rate(&e);
+        if (!rate_is_near(high, RATE)) {
+                printf("Attack missed: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) RATE, high);
+                goto fail;
+        }
+
+        for (i = 41; i <= 80; i++)
+                cap_update_at(&e, QLEN, LEN / 2, i * TICK);
+
+        if (cap_rate(&e) >= high) {
+                printf("Release did not decay: %" PRIu64 ".\n", cap_rate(&e));
+                goto fail;
+        }
+
+        if (cap_rate(&e) <= RATE / 2) {
+                printf("Release collapsed to %" PRIu64 ".\n", cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/* No window close within CAP_T_MIN of the last one. */
+static int test_cap_est_gate(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        cap_update_at(&e, QLEN, LEN, CAP_T_MIN);
+
+        for (i = 0; i < 5; i++)
+                cap_update_at(&e, QLEN, LEN, CAP_T_MIN + CAP_T_MIN / 2);
+
+        if (e.t_gate != CAP_T_MIN) {
+                printf("Window closed inside the gate.\n");
+                goto fail;
+        }
+
+        if (LOAD_RELAXED(&e.c_pkt) != 6) {
+                printf("Gated packets not counted.\n");
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+/*
+ * A link slow enough that CAP_N_MIN packets take longer than
+ * CAP_T_MAX to drain still publishes, as long as the sender keeps
+ * offering: only silence voids a window.
+ */
+static int test_cap_est_low_rate_publishes(void)
+{
+        struct cap_est e;
+        size_t         i;
+
+        TEST_START();
+
+        cap_clear(&e);
+
+        for (i = 1; i <= 20; i++)
+                cap_update_at(&e, QLEN, LEN, i * LOW_STEP);
+
+        if (!rate_is_near(cap_rate(&e), LOW_RATE)) {
+                printf("Low rate: exp %" PRIu64 ", got %" PRIu64 ".\n",
+                       (uint64_t) LOW_RATE, cap_rate(&e));
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+int cap_test(int     argc,
+             char ** argv)
+{
+        int ret = 0;
+
+        (void) argc;
+        (void) argv;
+
+        ret |= test_cap_est_clear();
+        ret |= test_cap_est_busy_window();
+        ret |= test_cap_est_idle_tolerated();
+        ret |= test_cap_est_mostly_idle_rejects();
+        ret |= test_cap_est_slow_link_extends();
+        ret |= test_cap_est_shaped_link();
+        ret |= test_cap_est_stale_discard();
+        ret |= test_cap_est_empty_start_no_raise();
+        ret |= test_cap_est_max_filter();
+        ret |= test_cap_est_gate();
+        ret |= test_cap_est_low_rate_publishes();
+
+        return ret;
+}