lib: Replace shim IPCPs with points of attachment

Removes the UDP and Ethernet shim IPCPs. The unicast and broadcast
IPCPs can now directly attach to a "legacy" socket. We adopt Saltzer's
Point-of-Attachment terminology, also advocated in Day's "Patterns in
Network Architecture". The "poa" component manages these
PoA's with one management thread, one link monitoring thread and one
thread per attached point.

For Ethernet PoA's the irm connect and enroll can resolve the
destination IPCP or Layer name with a broadcast name query over the
attached PoAs (first reply wins). UDP PoA's require a destination IP
address or FQDN.

attach to a local endpoint (required both server and client side):

 irm ipcp poa attach name a udp 10.0.0.1
 irm ipcp poa attach name a udp 10.0.0.1:3435
 irm ipcp poa attach name a udp [::1]:3435
 irm ipcp poa attach name a eth dev eth0
 irm ipcp poa attach name a eth dev eth0 ethertype 0xA000

release a PoA (refused while it carries a flow):

 irm ipcp poa detach name a udp 10.0.0.1:3435
 irm ipcp poa detach name a eth eth0

list an IPCP's PoAs:

 irm ipcp poa list name a

connect to a peer, by name or at an address:

 irm ipcp connect name b dst a
 irm ipcp connect name b dst a eth
 irm ipcp connect name b dst a eth dev eth0
 irm ipcp connect name b dst a udp 10.0.0.1:3435
 irm ipcp connect name b dst a udp peer.example.com:3435

disconnect by peer name, no address:

 irm ipcp disconnect name b dst a
 irm ipcp disconnect name b dst a component mgmt

enroll has the same shape as connect:

 irm ipcp enroll name b layer lr autobind
 irm ipcp enroll name b layer lr autobind eth dev eth0
 irm ipcp enroll name b layer lr autobind udp 10.0.0.1:3435

the IRMd config file attaches PoAs and names peers the same way:

 udp = [ "10.0.0.1", "10.0.0.1:3436" ]
 eth = [ "eth0", {dev="eth1", ethertype=0xA007} ]

 enrol={dst="LAN", eth={dev="eth0"}}
 conn=[{dst="lan3", eth={}}, {dst="lan4", udp="10.0.0.1:3435"}]

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

17 files changed, 6686 insertions, 142 deletions

diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt
index 3abf39d0..48064ce4 100644
--- a/src/lib/CMakeLists.txt
+++ b/src/lib/CMakeLists.txt
@@ -95,9 +95,9 @@ endif()
 
 if(HAVE_LIBURCU)
   target_link_libraries(ouroboros-common PRIVATE Urcu::Urcu)
-  # urcu headers require C99; override the global -std=c89 for this TU only.
-  set_source_files_properties(crypt/keyrot.c PROPERTIES
-    COMPILE_OPTIONS "-std=gnu99")
+  # urcu headers require C99; override the global -std=c89 for these TUs.
+  set_source_files_properties(crypt/keyrot.c dev.c poa/addr.c poa/eth.c
+    poa/udp.c PROPERTIES COMPILE_OPTIONS "-std=gnu99")
 endif()
 
 install(TARGETS ouroboros-common
@@ -107,8 +107,14 @@ install(TARGETS ouroboros-common
 set(SOURCE_FILES_DEV
   cep.c
   dev.c
+  poa/addr.c
+  poa/udp.c
 )
 
+if(HAVE_ETH)
+  list(APPEND SOURCE_FILES_DEV poa/eth.c)
+endif()
+
 add_library(ouroboros-dev SHARED
   ${SOURCE_FILES_DEV}
   ${CEP_PROTO_SRCS})
@@ -131,6 +137,11 @@ target_include_directories(ouroboros-dev
 
 target_link_libraries(ouroboros-dev PUBLIC ouroboros-common)
 
+if(HAVE_LIBURCU)
+  # dev.c calls urcu directly; do not rely on transitive linkage.
+  target_link_libraries(ouroboros-dev PRIVATE Urcu::Urcu)
+endif()
+
 install(TARGETS ouroboros-dev
   EXPORT OuroborosTargets
   LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR})
diff --git a/src/lib/config.h.in b/src/lib/config.h.in
index 4ebaa194..3c6985c3 100644
--- a/src/lib/config.h.in
+++ b/src/lib/config.h.in
@@ -79,6 +79,25 @@
 #define PROC_RES_FDS        @PROC_RES_FDS@
 #define PROC_MAX_FQUEUES    @PROC_MAX_FQUEUES@
 
+/* Flow endpoints */
+#define POA_MGMT_FRAME_SIZE  @POA_MGMT_FRAME_SIZE@
+#define POA_MGMT_SND_TIMEO   @POA_MGMT_SND_TIMEO@
+#define POA_MAX_POAS         @POA_MAX_POAS@
+#define POA_UDP_MPL          @POA_UDP_MPL@
+#define POA_UDP4_MTU         @POA_UDP4_MTU@
+#define POA_UDP6_MTU         @POA_UDP6_MTU@
+#define POA_UDP_RD_BUF       @POA_UDP_RD_BUF@
+#cmakedefine                HAVE_ETH
+#cmakedefine                HAVE_RAW_SOCKETS
+#cmakedefine                HAVE_BPF
+#cmakedefine                HAVE_NETMAP
+#cmakedefine                POA_ETH_QDISC_BYPASS
+#define POA_ETH_MPL          @POA_ETH_MPL@
+#define POA_ETH_SNDBUF       @POA_ETH_SNDBUF@
+#define POA_ETH_RCVBUF       @POA_ETH_RCVBUF@
+#define POA_ETH_LO_MTU       @POA_ETH_LO_MTU@
+#define POA_ETH_RD_BUF       @POA_ETH_RD_BUF@
+
 /* Default Delta-t parameters */
 #cmakedefine                FRCT_LINUX_RTT_ESTIMATOR
 #define DELT_A              (@DELTA_T_ACK@)                  /* ms */
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 3fb8d831..eb706691 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -28,6 +28,9 @@
 
 #include "config.h"
 #include "ssm.h"
+#include "poa/poa.h"
+
+#define OUROBOROS_PREFIX "libouroboros"
 
 #include <ouroboros/atomics.h>
 #include <ouroboros/bitmap.h>
@@ -45,6 +48,7 @@
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/list.h>
 #include <ouroboros/local-dev.h>
+#include <ouroboros/logs.h>
 #include <ouroboros/np1_flow.h>
 #include <ouroboros/pthread.h>
 #include <ouroboros/random.h>
@@ -63,8 +67,10 @@
 #ifdef HAVE_LIBGCRYPT
 #include <gcrypt.h>
 #endif
+#include <arpa/inet.h>
 #include <stdarg.h>
 #include <stdbool.h>
+#include <stddef.h>
 #include <inttypes.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -98,8 +104,8 @@ struct flow {
         ssize_t               part_idx;
 
         struct crypt_ctx *    crypt;
-        int                   headsz;         /* Selector                  */
-        int                   tailsz;         /* Tag + CRC                 */
+        int                   headsz;         /* Selector                   */
+        int                   tailsz;         /* Tag + CRC                  */
 
         struct timespec       rk_grace;       /* TX-promote deadline       */
         struct timespec       rk_attempt;     /* Last re-key attempt       */
@@ -116,6 +122,8 @@ struct flow {
         struct timespec       rcv_timeo;
 
         struct frcti *        frcti;
+
+        struct poa_flow *     poa;   /* NULL for shared memory flows */
 };
 
 struct flow_set {
@@ -696,6 +704,9 @@ static void do_flow_fini(int fd)
 {
         assert(fd >= 0 && fd < PROC_MAX_FLOWS);
 
+        if (proc.flows[fd].poa != NULL)
+                poa_flow_detach(proc.flows[fd].poa);
+
         if (proc.flows[fd].frcti != NULL)
                 frcti_destroy(proc.flows[fd].frcti);
 
@@ -750,9 +761,11 @@ static __inline__ size_t flow_user_mtu(const struct flow * flow,
         return raw > hdr ? raw - hdr : 0;
 }
 
+/* A PoA flow transmits on its own socket; it has no tx ring. */
 static int flow_init(struct flow_info * info,
                      struct crypt_sk *  sk,
-                     time_t             rtt_hint)
+                     time_t             rtt_hint,
+                     struct poa_flow *  pf)
 {
         struct timespec now;
         struct flow *   flow;
@@ -777,13 +790,15 @@ static int flow_init(struct flow_info * info,
         if (flow->rx_rb == NULL)
                 goto fail_rx_rb;
 
-        flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id);
-        if (flow->tx_rb == NULL)
-                goto fail_tx_rb;
+        if (pf == NULL) {
+                flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id);
+                if (flow->tx_rb == NULL)
+                        goto fail_tx_rb;
 
-        flow->set = ssm_flow_set_open(info->n_1_pid);
-        if (flow->set == NULL)
-                goto fail_set;
+                flow->set = ssm_flow_set_open(info->n_1_pid);
+                if (flow->set == NULL)
+                        goto fail_set;
+        }
 
         flow->oflags   = FLOWFDEFAULT;
         flow->part_idx = NO_PART;
@@ -792,11 +807,14 @@ static int flow_init(struct flow_info * info,
         flow->crypt    = NULL;
         flow->headsz   = 0;
         flow->tailsz   = 0;
+        flow->poa      = pf;
 
         if (IS_ENCRYPTED(sk)) {
                 flow->crypt = crypt_create_ctx(sk);
-                if (flow->crypt == NULL)
+                if (flow->crypt == NULL) {
+                        err = -ECRYPT;
                         goto fail_crypt;
+                }
                 flow->headsz = crypt_get_headsz(flow->crypt);
                 flow->tailsz = crypt_get_tagsz(flow->crypt);
         }
@@ -815,6 +833,9 @@ static int flow_init(struct flow_info * info,
 
         proc.id_to_fd[info->id].fd = fd;
 
+        if (pf != NULL)
+                poa_flow_attach(pf, info->id, flow->rx_rb);
+
         flow_set_state(&proc.id_to_fd[info->id], FLOW_ALLOCATED);
 
         pthread_rwlock_unlock(&proc.lock);
@@ -824,9 +845,11 @@ static int flow_init(struct flow_info * info,
  fail_frcti:
         crypt_destroy_ctx(flow->crypt);
  fail_crypt:
-        ssm_flow_set_close(flow->set);
+        if (flow->set != NULL)
+                ssm_flow_set_close(flow->set);
  fail_set:
-        ssm_rbuff_close(flow->tx_rb);
+        if (flow->tx_rb != NULL)
+                ssm_rbuff_close(flow->tx_rb);
  fail_tx_rb:
         ssm_rbuff_close(flow->rx_rb);
  fail_rx_rb:
@@ -1077,14 +1100,20 @@ static void fini(void)
 __attribute__((section(INIT_SECTION))) __typeof__(init) * __init = init;
 __attribute__((section(FINI_SECTION))) __typeof__(fini) * __fini = fini;
 
+/*
+ * A PoA flow is announced by its peer before the accept; from the
+ * reply on, the peer may transmit, so the flow must be able to
+ * receive.
+ */
 int flow_accept(qosspec_t *             qs,
                 const struct timespec * timeo)
 {
         struct flow_info  flow;
-        struct crypt_sk crypt;
+        struct crypt_sk   crypt;
         uint8_t           buf[SOCK_BUF_SIZE];
         buffer_t          msg = {SOCK_BUF_SIZE, buf};
         uint8_t           key[SYMMKEYSZ];
+        struct poa_flow * pf;
         int               fd;
         int               err;
 
@@ -1116,8 +1145,13 @@ int flow_accept(qosspec_t *             qs,
         if (err < 0)
                 return err;
 
-        /* No RTT in accept; rtt_hint=0 bootstraps from first ACK. */
-        fd = flow_init(&flow, &crypt, 0);
+        pf = poa_flow_take_pending(flow.id);
+
+        fd = flow_init(&flow, &crypt, 0, pf);
+        if (fd >= 0)
+                poa_flow_ready(pf);
+        else if (pf != NULL)
+                poa_flow_detach(pf);
 
         crypt_secure_clear(key, SYMMKEYSZ);
 
@@ -1173,7 +1207,7 @@ int flow_alloc(const char *            dst,
         if (err < 0)
                 return err;
 
-        fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0));
+        fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), NULL);
 
         crypt_secure_clear(key, SYMMKEYSZ);
 
@@ -1214,7 +1248,7 @@ int flow_join(const char *            dst,
         if (err < 0)
                 return err;
 
-        fd = flow_init(&flow, &crypt, 0);
+        fd = flow_init(&flow, &crypt, 0, NULL);
 
         crypt_secure_clear(key, SYMMKEYSZ);
 
@@ -1275,7 +1309,8 @@ int flow_dealloc(int fd)
 
         pthread_cleanup_push(__cleanup_rwlock_unlock, &proc.lock);
 
-        ssm_rbuff_fini(flow->tx_rb);
+        if (flow->tx_rb != NULL)
+                ssm_rbuff_fini(flow->tx_rb);
 
         pthread_cleanup_pop(true);
 
@@ -1428,7 +1463,11 @@ int fccntl(int fd,
                 break;
         case FLOWGTXQLEN:
                 qlen  = va_arg(l, size_t *);
-                *qlen = ssm_rbuff_queued(flow->tx_rb);
+
+                if (flow->poa != NULL)
+                        *qlen = poa_flow_qpkts(flow->poa);
+                else
+                        *qlen = ssm_rbuff_queued(flow->tx_rb);
                 break;
         case FLOWGMTU:
                 maxp = va_arg(l, size_t *);
@@ -1456,16 +1495,18 @@ int fccntl(int fd,
 
                 if (flow->oflags & FLOWFDOWN) {
                         ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN);
-                        ssm_rbuff_set_bits(flow->tx_rb, RB_FLOWDOWN);
-                        ssm_flow_set_notify(flow->set,
-                                            flow->info.id,
-                                            FLOW_DOWN);
+                        if (flow->tx_rb != NULL)
+                                ssm_rbuff_set_bits(flow->tx_rb, RB_FLOWDOWN);
+                        if (flow->set != NULL)
+                                ssm_flow_set_notify(flow->set, flow->info.id,
+                                                    FLOW_DOWN);
                 } else {
                         ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN);
-                        ssm_rbuff_clr_bits(flow->tx_rb, RB_FLOWDOWN);
-                        ssm_flow_set_notify(flow->set,
-                                            flow->info.id,
-                                            FLOW_UP);
+                        if (flow->tx_rb != NULL)
+                                ssm_rbuff_clr_bits(flow->tx_rb, RB_FLOWDOWN);
+                        if (flow->set != NULL)
+                                ssm_flow_set_notify(flow->set, flow->info.id,
+                                                    FLOW_UP);
                 }
 
                 break;
@@ -1599,6 +1640,9 @@ static int flow_tx_spb(struct flow *        flow,
                         goto enomem;
         }
 
+        if (flow->poa != NULL)
+                return poa_flow_tx(flow->poa, spb, block, abstime);
+
         if (!block)
                 ret = ssm_rbuff_write(flow->tx_rb, idx);
         else
@@ -2450,7 +2494,7 @@ int np1_flow_alloc(pid_t n_pid,
         /* np1 flow: n_1_pid is the upper. */
         flow.n_1_pid = n_pid;
 
-        return flow_init(&flow, &crypt, 0);
+        return flow_init(&flow, &crypt, 0, NULL);
 }
 
 int np1_flow_dealloc(int    flow_id,
@@ -2584,7 +2628,7 @@ int ipcp_flow_req_arr(const buffer_t * dst,
 
         crypt.nid = NID_undef;
 
-        return flow_init(&flow, &crypt, 0);
+        return flow_init(&flow, &crypt, 0, NULL);
 }
 
 int ipcp_flow_update_arr(int              flow_id,
@@ -2853,11 +2897,13 @@ int ipcp_flow_fini(int fd)
         }
 
         ssm_rbuff_set_bits(proc.flows[fd].rx_rb, RB_FLOWDOWN);
-        ssm_rbuff_set_bits(proc.flows[fd].tx_rb, RB_FLOWDOWN);
 
-        ssm_flow_set_notify(proc.flows[fd].set,
-                            proc.flows[fd].info.id,
-                            FLOW_DEALLOC);
+        if (proc.flows[fd].tx_rb != NULL)
+                ssm_rbuff_set_bits(proc.flows[fd].tx_rb, RB_FLOWDOWN);
+
+        if (proc.flows[fd].set != NULL)
+                ssm_flow_set_notify(proc.flows[fd].set, proc.flows[fd].info.id,
+                                    FLOW_DEALLOC);
 
         rx_rb = proc.flows[fd].rx_rb;
 
@@ -2892,9 +2938,30 @@ size_t ipcp_flow_queued(int fd)
         assert(fd >= 0 && fd < PROC_MAX_FLOWS);
         assert(proc.flows[fd].info.id >= 0);
 
+        if (proc.flows[fd].poa != NULL)
+                return poa_flow_qlen(proc.flows[fd].poa);
+
         return ssm_rbuff_queued(proc.flows[fd].tx_rb);
 }
 
+int ipcp_flow_queue_id(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 fd;
+
+        /* 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;
+
+        return PROC_MAX_FLOWS + qid;
+}
+
 int local_flow_transfer(int               src_fd,
                         int               dst_fd,
                         struct ssm_pool * src_pool,
@@ -2960,3 +3027,5 @@ int local_flow_transfer(int               src_fd,
 
         return ret;
 }
+
+#include "poa/poa.c"
diff --git a/src/lib/irm.c b/src/lib/irm.c
index c62701aa..5d53841f 100644
--- a/src/lib/irm.c
+++ b/src/lib/irm.c
@@ -118,6 +118,8 @@ int irm_bootstrap_ipcp(pid_t                      pid,
         msg.has_pid = true;
         msg.pid     = pid;
         msg.conf    = ipcp_config_s_to_msg(conf);
+        if (msg.conf == NULL)
+                return -ENOMEM;
 
         recv_msg = send_recv_irm_msg(&msg);
         ipcp_config_msg__free_unpacked(msg.conf, NULL);
@@ -135,10 +137,11 @@ int irm_bootstrap_ipcp(pid_t                      pid,
         return ret;
 }
 
-int irm_connect_ipcp(pid_t        pid,
-                     const char * dst,
-                     const char * component,
-                     qosspec_t    qs)
+int irm_connect_ipcp(pid_t                   pid,
+                     const char *            dst,
+                     const char *            component,
+                     qosspec_t               qs,
+                     const struct poa_addr * addr)
 {
         irm_msg_t     msg = IRM_MSG__INIT;
         irm_msg_t *   recv_msg;
@@ -152,7 +155,18 @@ int irm_connect_ipcp(pid_t        pid,
         msg.pid       = pid;
         msg.qosspec   = qos_spec_s_to_msg(&qs);
 
+        if (addr != NULL) {
+                msg.peer = poa_addr_s_to_msg(addr);
+                if (msg.peer == NULL) {
+                        qosspec_msg__free_unpacked(msg.qosspec, NULL);
+                        return -ENOMEM;
+                }
+        }
+
         recv_msg = send_recv_irm_msg(&msg);
+
+        if (msg.peer != NULL)
+                poa_addr_msg__free_unpacked(msg.peer, NULL);
         qosspec_msg__free_unpacked(msg.qosspec, NULL);
 
         if (recv_msg == NULL)
@@ -245,8 +259,99 @@ ssize_t irm_list_ipcps(struct ipcp_list_info ** ipcps)
         return nr;
 }
 
-int irm_enroll_ipcp(pid_t        pid,
-                    const char * dst)
+static int irm_attach_detach(pid_t                   pid,
+                             const struct poa_spec * poa,
+                             int                     code)
+{
+        irm_msg_t   msg = IRM_MSG__INIT;
+        irm_msg_t * recv_msg;
+        int         ret;
+
+        if (pid == -1 || poa == NULL)
+                return -EINVAL;
+
+        msg.code     = code;
+        msg.has_pid  = true;
+        msg.pid      = pid;
+        msg.poa = poa_spec_s_to_msg(poa);
+        if (msg.poa == NULL)
+                return -EINVAL;
+
+        recv_msg = send_recv_irm_msg(&msg);
+
+        poa_spec_msg__free_unpacked(msg.poa, NULL);
+
+        if (recv_msg == NULL)
+                return -EIRMD;
+
+        if (!recv_msg->has_result) {
+                irm_msg__free_unpacked(recv_msg, NULL);
+                return -EIRMD;
+        }
+
+        ret = recv_msg->result;
+
+        irm_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+ssize_t irm_list_poas(pid_t              pid,
+                      struct poa_spec ** poas)
+{
+        irm_msg_t   msg = IRM_MSG__INIT;
+        irm_msg_t * recv_msg;
+        size_t      nr;
+        size_t      i;
+
+        if (pid == -1 || poas == NULL)
+                return -EINVAL;
+
+        *poas = NULL;
+
+        msg.code    = IRM_MSG_CODE__IRM_LIST_POAS;
+        msg.has_pid = true;
+        msg.pid     = pid;
+
+        recv_msg = send_recv_irm_msg(&msg);
+        if (recv_msg == NULL)
+                return -EIRMD;
+
+        nr = recv_msg->n_poas;
+        if (nr == 0) {
+                irm_msg__free_unpacked(recv_msg, NULL);
+                return 0;
+        }
+
+        *poas = malloc(nr * sizeof(**poas));
+        if (*poas == NULL) {
+                irm_msg__free_unpacked(recv_msg, NULL);
+                return -ENOMEM;
+        }
+
+        for (i = 0; i < nr; i++)
+                (*poas)[i] = poa_spec_msg_to_s(recv_msg->poas[i]);
+
+        irm_msg__free_unpacked(recv_msg, NULL);
+
+        return (ssize_t) nr;
+}
+
+int irm_attach_ipcp(pid_t                   pid,
+                    const struct poa_spec * poa)
+{
+        return irm_attach_detach(pid, poa, IRM_MSG_CODE__IRM_ATTACH_IPCP);
+}
+
+int irm_detach_ipcp(pid_t                   pid,
+                    const struct poa_spec * poa)
+{
+        return irm_attach_detach(pid, poa, IRM_MSG_CODE__IRM_DETACH_IPCP);
+}
+
+int irm_enroll_ipcp(pid_t                   pid,
+                    const char *            dst,
+                    const struct poa_addr * addr)
 {
         irm_msg_t   msg      = IRM_MSG__INIT;
         irm_msg_t * recv_msg = NULL;
@@ -260,7 +365,17 @@ int irm_enroll_ipcp(pid_t        pid,
         msg.pid          = pid;
         msg.dst          = (char *) dst;
 
+        if (addr != NULL) {
+                msg.peer = poa_addr_s_to_msg(addr);
+                if (msg.peer == NULL)
+                        return -ENOMEM;
+        }
+
         recv_msg = send_recv_irm_msg(&msg);
+
+        if (msg.peer != NULL)
+                poa_addr_msg__free_unpacked(msg.peer, NULL);
+
         if (recv_msg == NULL)
                 return -EIRMD;
 
diff --git a/src/lib/pb/ipcp.proto b/src/lib/pb/ipcp.proto
index afee4f91..298a15e2 100644
--- a/src/lib/pb/ipcp.proto
+++ b/src/lib/pb/ipcp.proto
@@ -40,6 +40,9 @@ enum ipcp_msg_code {
         IPCP_DISCONNECT      = 11;
         IPCP_REPLY           = 12;
         IPCP_FLOW_UPDATE     = 13;
+        IPCP_ATTACH          = 14;
+        IPCP_DETACH          = 15;
+        IPCP_LIST_POAS  = 16;
 }
 
 message ipcp_msg {
@@ -58,4 +61,8 @@ message ipcp_msg {
         optional sint32 mpl                = 13; /* MPL in ms. */
         optional int32 result              = 14;
         optional uint32 uid                = 15; /* 0 = GSPP, >0 = PUP uid */
+        optional poa_addr_msg peer          = 16; /* peer PoA address       */
+        optional bool is_poa               = 17; /* flow rides a PoA       */
+        optional poa_spec_msg poa          = 18; /* PoA to attach          */
+        repeated poa_spec_msg poas         = 19; /* PoAs attached          */
 }
diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto
index a111b6d8..f6d313ac 100644
--- a/src/lib/pb/ipcp_config.proto
+++ b/src/lib/pb/ipcp_config.proto
@@ -66,29 +66,32 @@ message uni_config_msg {
         required uint32 cong_avoid     = 4;
 }
 
-message eth_config_msg {
+message eth_poa_msg {
         required string dev       = 1;
         required uint32 ethertype = 2;
+        optional bytes  mac       = 3;
 }
 
-message udp4_config_msg {
-        required uint32 ip_addr  = 1;
-        required uint32 port     = 2;
-        required uint32 dns_addr = 3; /* set to 0 if unused */
+message udp4_poa_msg {
+        required uint32 ip_addr = 1;
+        required uint32 port    = 2;
 }
 
-message udp6_config_msg {
-        required bytes  ip_addr  = 1;
-        required uint32 port     = 2;
-        required bytes  dns_addr = 3; /* set to NULL if unused */
+message udp6_poa_msg {
+        required bytes  ip_addr = 1;
+        required uint32 port    = 2;
 }
 
 
+/* Exactly one field is set; presence is the PoA type. */
+message poa_spec_msg {
+        optional udp4_poa_msg udp4 = 1;
+        optional udp6_poa_msg udp6 = 2;
+        optional eth_poa_msg  eth  = 3;
+}
+
 message ipcp_config_msg {
         required layer_info_msg layer_info =  1;
         required uint32 ipcp_type          =  2;
         optional uni_config_msg unicast    =  3;
-        optional udp4_config_msg udp4      =  4;
-        optional udp6_config_msg udp6      =  5;
-        optional eth_config_msg eth        =  6;
 }
diff --git a/src/lib/pb/irm.proto b/src/lib/pb/irm.proto
index f54bc9ea..8f594fa3 100644
--- a/src/lib/pb/irm.proto
+++ b/src/lib/pb/irm.proto
@@ -55,6 +55,12 @@ enum irm_msg_code {
         IRM_REPLY             = 27;
         IRM_FLOW_UPDATE       = 28;
         IPCP_FLOW_UPDATE_ARR  = 29;
+        IRM_POA_FLOW_ALLOC     = 30;
+        IRM_POA_FLOW_ALLOC_R   = 31;
+        IPCP_POA_FLOW_REQ_ARR  = 32;
+        IRM_ATTACH_IPCP       = 33;
+        IRM_DETACH_IPCP       = 34;
+        IRM_LIST_POAS    = 35;
 }
 
 message timespec_msg {
@@ -101,4 +107,7 @@ message irm_msg {
         optional uint32 generation       = 28; /* re-key batch generation */
         optional bool   rekey            = 29; /* re-key watermark trigger */
         optional bool   rk_initiator     = 30; /* re-key proof-holder side */
+        optional poa_addr_msg peer       = 32; /* peer PoA address     */
+        optional poa_spec_msg poa        = 33; /* PoA to attach        */
+        repeated poa_spec_msg poas       = 34; /* PoAs attached        */
 }
diff --git a/src/lib/pb/model.proto b/src/lib/pb/model.proto
index 4c1564a5..e00f3e12 100644
--- a/src/lib/pb/model.proto
+++ b/src/lib/pb/model.proto
@@ -58,6 +58,19 @@ message layer_info_msg {
         required uint32 dir_hash_algo = 2;
 }
 
+/* Address of a flow PoA peer. */
+message poa_addr_msg {
+        required uint32 type      = 1; /* enum poa_type */
+        optional uint32 ip4       = 2;
+        optional bytes  ip6       = 3;
+        optional uint32 port      = 4;
+        optional bytes  mac       = 5;
+        optional uint32 ethertype = 6;
+        optional string dev       = 7; /* our device, eth only    */
+        optional string hostname  = 8; /* IRMd resolves, UDP only */
+        optional bytes  src_mac   = 9; /* our MAC, eth only       */
+}
+
 message ipcp_info_msg {
         required uint32 type  = 1;
         required string name  = 2;
diff --git a/src/lib/poa/addr.c b/src/lib/poa/addr.c
new file mode 100644
index 00000000..d8811244
--- /dev/null
+++ b/src/lib/poa/addr.c
@@ -0,0 +1,142 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - addresses and management messages
+ *
+ *    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/.
+ */
+
+#if defined(__linux__) || defined(__CYGWIN__)
+#ifndef _DEFAULT_SOURCE /* Test include source */
+#define _DEFAULT_SOURCE
+#endif
+#endif
+
+#include "config.h"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+
+#include "poa.h"
+
+#include <arpa/inet.h>
+#include <stdio.h>
+#include <string.h>
+
+int poa_addr_cmp(const struct poa_addr * a,
+                 const struct poa_addr * b)
+{
+        if (a->type != b->type)
+                return 1;
+
+        switch (a->type) {
+        case POA_UDP4:
+                if (a->udp4.port != b->udp4.port)
+                        return 1;
+                return memcmp(&a->udp4.ip_addr, &b->udp4.ip_addr,
+                              sizeof(a->udp4.ip_addr));
+        case POA_UDP6:
+                if (a->udp6.port != b->udp6.port)
+                        return 1;
+                return memcmp(&a->udp6.ip_addr, &b->udp6.ip_addr,
+                              sizeof(a->udp6.ip_addr));
+        case POA_ETH:
+                if (a->eth.dst.ethertype != b->eth.dst.ethertype)
+                        return 1;
+                return memcmp(a->eth.dst.mac, b->eth.dst.mac, POA_MAC_SIZE);
+        default:
+                return 1;
+        }
+}
+
+/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */
+int poa_addr_name(const struct poa_addr * a,
+                  char *                  buf,
+                  size_t                  len)
+{
+        char ip[INET6_ADDRSTRLEN];
+        int  ret;
+
+        switch (a->type) {
+        case POA_UDP4:
+                if (inet_ntop(AF_INET, &a->udp4.ip_addr,
+                              ip, sizeof(ip)) == NULL)
+                        return -EINVAL;
+                ret = snprintf(buf, len, "udp4.%s.%u", ip, a->udp4.port);
+                break;
+        case POA_UDP6:
+                if (inet_ntop(AF_INET6, &a->udp6.ip_addr,
+                              ip, sizeof(ip)) == NULL)
+                        return -EINVAL;
+                ret = snprintf(buf, len, "udp6.%s.%u", ip, a->udp6.port);
+                break;
+        case POA_ETH:
+                ret = snprintf(buf, len, "eth.%s.%04X",
+                               a->eth.src.dev, a->eth.src.ethertype);
+                break;
+        default:
+                return -EINVAL;
+        }
+
+        if (ret < 0 || (size_t) ret >= len)
+                return -EMSGSIZE;
+
+        return 0;
+}
+
+void poa_mgmt_msg_ser(struct poa_mgmt_msg * msg,
+                      uint8_t               code,
+                      uint32_t              s_eid,
+                      uint32_t              d_eid,
+                      qosspec_t             qs,
+                      int                   response,
+                      size_t                data_len)
+{
+        memset(msg, 0, sizeof(*msg));
+
+        msg->code         = code;
+        msg->s_eid        = hton32(s_eid);
+        msg->d_eid        = hton32(d_eid);
+        msg->response     = hton32(response);
+        msg->data_len     = hton16((uint16_t) data_len);
+
+        if (code != POA_FLOW_REQ)
+                return;
+
+        msg->bandwidth    = hton64(qs.bandwidth);
+        msg->delay        = hton32(qs.delay);
+        msg->loss         = hton32(qs.loss);
+        msg->ber          = hton32(qs.ber);
+        msg->max_gap      = hton32(qs.max_gap);
+        msg->timeout      = hton32(qs.timeout);
+        msg->availability = qs.availability;
+        msg->service      = qs.service;
+}
+
+void poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg,
+                      qosspec_t *                 qs)
+{
+        qs->bandwidth    = ntoh64(msg->bandwidth);
+        qs->delay        = ntoh32(msg->delay);
+        qs->loss         = ntoh32(msg->loss);
+        qs->ber          = ntoh32(msg->ber);
+        qs->max_gap      = ntoh32(msg->max_gap);
+        qs->timeout      = ntoh32(msg->timeout);
+        qs->availability = msg->availability;
+        qs->service      = msg->service;
+}
+
diff --git a/src/lib/poa/eth.c b/src/lib/poa/eth.c
new file mode 100644
index 00000000..5e959ef6
--- /dev/null
+++ b/src/lib/poa/eth.c
@@ -0,0 +1,1987 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - Ethernet transport
+ *
+ *    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/.
+ */
+
+#if defined(__APPLE__)
+#define _BSD_SOURCE
+#define _DARWIN_C_SOURCE
+#elif defined(__FreeBSD__)
+#define __BSD_VISIBLE 1
+#elif 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"
+
+#define OUROBOROS_PREFIX "poa-eth"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/hash.h>
+#include <ouroboros/logs.h>
+#include <ouroboros/pthread.h>
+#include <ouroboros/sockets.h>
+#include <ouroboros/time.h>
+
+#include "poa.h"
+
+#ifdef HAVE_RAW_SOCKETS
+#include <net/if.h>
+#include <netinet/in.h>
+#include <linux/if_ether.h>
+#include <linux/if_packet.h>
+#include <linux/netlink.h>
+#include <linux/gen_stats.h>
+#include <linux/pkt_sched.h>
+#include <linux/rtnetlink.h>
+#include <linux/sockios.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <ifaddrs.h>
+#elif defined(HAVE_BPF)
+#include <net/bpf.h>
+#include <net/if.h>
+#include <net/if_dl.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <fcntl.h>
+#include <ifaddrs.h>
+#elif defined(HAVE_NETMAP)
+#define NETMAP_WITH_LIBS
+#include <net/netmap_user.h>
+#include <net/if.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+
+#include <poll.h>
+#ifndef __linux__
+#include <net/if_dl.h>
+#include <ifaddrs.h>
+#endif
+#endif
+
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define ETH_TYPE_LEN_SIZE sizeof(uint16_t)
+#define ETH_HDR_SIZE      (2 * POA_MAC_SIZE + ETH_TYPE_LEN_SIZE)
+#define POA_HDR_SIZE      (sizeof(struct poa_hdr))
+#define ETH_HDR_TOT_SIZE  (ETH_HDR_SIZE + POA_HDR_SIZE)
+
+#define BPF_DEV_MAX       256  /* /dev/bpfN units to try */
+/* Wait for the link to come back before reading it again. */
+#define ETH_DOWN_TIMEO    100  /* ms */
+/* Budget for a qdisc dump; the send path must not wait on netlink. */
+#define POA_QDISC_TIMEO   5    /* ms */
+
+#ifndef ETH_MAX_MTU        /* In if_ether.h as of Linux 4.10. */
+#define ETH_MAX_MTU       0xFFFFU
+#endif
+
+struct poa_hdr {
+        uint16_t eid;
+        uint16_t len;
+        uint8_t  hcs;
+} __attribute__((packed));
+
+struct eth_hdr {
+        uint8_t        dst[POA_MAC_SIZE];
+        uint8_t        src[POA_MAC_SIZE];
+        uint16_t       ethertype;
+        struct poa_hdr poa;
+} __attribute__((packed));
+
+
+struct eth_priv {
+        struct poa *       poa;
+
+        int                s_fd;      /* raw socket or bpf device */
+        uint16_t           ethertype; /* network order            */
+        uint8_t            hw_addr[POA_MAC_SIZE];
+        uint32_t           mtu;       /* device payload capacity */
+        int                ifindex;   /* link id; 0 where unknown */
+
+        pthread_t          reader;
+        bool               running;
+#ifdef HAVE_RAW_SOCKETS
+        struct sockaddr_ll device;
+#elif defined(HAVE_BPF)
+        size_t             blen;      /* bpf device buffer size */
+#elif defined(HAVE_NETMAP)
+        struct nm_desc *   nmd;
+        struct pollfd      poll_in;
+        struct pollfd      poll_out;
+#endif
+        /* The kernel zeroes its counters on read, so accumulate. */
+        size_t             kern_rcv;
+        size_t             kern_drp;
+#ifdef HAVE_RAW_SOCKETS
+        /* Qdisc depth: netlink descriptor, its try-lock and cache. */
+        int                nl_fd;
+        uint8_t            nl_busy;
+        size_t             nl_pkt;
+        size_t             nl_byt;
+        uint64_t           nl_time;
+#endif
+};
+
+struct eth_query {
+        struct list_head next;
+        uint8_t          hash[POA_QUERY_HLEN];
+        /* The resolve's constraints; replies must satisfy them. */
+        uint16_t         ethertype;
+        char             c_dev[DEV_NAME_SIZE + 1];
+        uint8_t          c_mac[POA_MAC_SIZE];
+        bool             replied;
+        char             dev[DEV_NAME_SIZE + 1];
+        uint8_t          src_mac[POA_MAC_SIZE];
+        uint8_t          mac[POA_MAC_SIZE];
+        uint16_t         r_ethertype;
+};
+
+static struct {
+        pthread_once_t  once;
+
+        struct llist    pending;
+        pthread_mutex_t mtx;
+        pthread_cond_t  cond;
+} queries = { .once = PTHREAD_ONCE_INIT };
+
+static void eth_hdr_ser(const struct eth_priv * priv,
+                        struct eth_hdr *        hdr,
+                        const uint8_t *         dst,
+                        uint32_t                eid,
+                        size_t                  len)
+{
+        memcpy(hdr->dst, dst, POA_MAC_SIZE);
+        memcpy(hdr->src, priv->hw_addr, POA_MAC_SIZE);
+
+        hdr->ethertype = priv->ethertype;
+        hdr->poa.eid   = htons((uint16_t) eid);
+        hdr->poa.len   = htons((uint16_t) len);
+        mem_hash(HASH_CRC8, &hdr->poa.hcs,
+                 (uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t));
+}
+
+/* Oriented from us to the sender: our PoA, then their MAC. */
+static void frame_to_addr(const struct eth_priv * priv,
+                          const struct eth_hdr *  hdr,
+                          struct poa_addr *       addr)
+{
+        memset(addr, 0, sizeof(*addr));
+
+        addr->type = priv->poa->type;
+
+        addr->eth.src = priv->poa->local.eth.src;
+
+        memcpy(addr->eth.dst.mac, hdr->src, POA_MAC_SIZE);
+
+        addr->eth.dst.ethertype = ntohs(priv->ethertype);
+}
+
+static bool frame_is_for_us(const struct eth_priv * priv,
+                            const uint8_t *         dst)
+{
+        static const uint8_t bc[POA_MAC_SIZE] =
+                { 0xff, 0xff, 0xff, 0xff, 0xff, 0xff };
+
+        if (memcmp(dst, priv->hw_addr, POA_MAC_SIZE) == 0)
+                return true;
+
+        return memcmp(dst, bc, POA_MAC_SIZE) == 0;
+}
+
+/*
+ * Parse and validate a frame; on success *eid and *plen locate the
+ * payload. Errors are silent: anyone can spray an interface.
+ */
+static int frame_parse(const struct eth_priv * priv,
+                       const uint8_t *         buf,
+                       size_t                  len,
+                       uint32_t *              eid,
+                       size_t *                plen)
+{
+        const struct eth_hdr * hdr = (const struct eth_hdr *) buf;
+        uint8_t                hcs;
+
+        if (len < ETH_HDR_TOT_SIZE)
+                return -1;
+
+        if (!frame_is_for_us(priv, hdr->dst))
+                return -1;
+
+        if (hdr->ethertype != priv->ethertype)
+                return -1;
+
+        mem_hash(HASH_CRC8, &hcs,
+                 (const uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t));
+
+        if (hcs != hdr->poa.hcs)
+                return -1;
+
+        *eid  = ntohs(hdr->poa.eid);
+
+        *plen = ntohs(hdr->poa.len);
+        if (*plen > len - ETH_HDR_TOT_SIZE)
+                return -1;
+
+        return 0;
+}
+
+#define ETH_QUERY_TIMEO   1900 /* ms total budget               */
+#define ETH_QUERY_RETRIES 3    /* retransmits, 4 attempts total */
+
+/* A resolve in progress; a reply fills every matching entry. */
+static void queries_init(void)
+{
+        pthread_condattr_t cattr;
+
+        llist_init(&queries.pending);
+
+        pthread_mutex_init(&queries.mtx, NULL);
+
+        pthread_condattr_init(&cattr);
+#ifndef __APPLE__
+        pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+        pthread_cond_init(&queries.cond, &cattr);
+        pthread_condattr_destroy(&cattr);
+}
+
+static void eth_query_cleanup(void * o)
+{
+        struct eth_query * q = (struct eth_query *) o;
+
+        pthread_mutex_lock(&queries.mtx);
+        llist_del(&q->next, &queries.pending);
+        pthread_mutex_unlock(&queries.mtx);
+}
+
+static void eth_query_arr(struct poa *            poa,
+                          const struct poa_addr * src,
+                          const uint8_t *         hash)
+{
+        uint8_t               buf[sizeof(struct poa_mgmt_msg)
+                                   + POA_QUERY_HLEN];
+        struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+
+        if (!poa_has_name(hash))
+                return;
+
+        poa_mgmt_msg_ser(msg, POA_NAME_REPLY, 0, 0, qos_raw, 0, POA_QUERY_HLEN);
+
+        memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN);
+
+        if (poa->ops->poa_send_mgmt(poa, src, buf, sizeof(buf)) < 0)
+                return; /* the requester retransmits */
+
+        POA_STAT_BUMP(poa, rep_tx);
+}
+
+static bool mac_is_zero(const uint8_t * mac)
+{
+        static const uint8_t zero[POA_MAC_SIZE] = { 0 };
+
+        return memcmp(mac, zero, POA_MAC_SIZE) == 0;
+}
+
+static bool eth_query_is_match(const struct eth_query * q,
+                               const struct poa *       poa)
+{
+        if (q->ethertype != 0 &&
+            q->ethertype != poa->local.eth.src.ethertype)
+                return false;
+
+        if (q->c_dev[0] != '\0' &&
+            strcmp(q->c_dev, poa->local.eth.src.dev) != 0)
+                return false;
+
+        if (mac_is_zero(q->c_mac))
+                return true;
+
+        return memcmp(q->c_mac, poa->local.eth.src.mac, POA_MAC_SIZE) == 0;
+}
+
+static void eth_reply_arr(struct poa *            poa,
+                          const struct poa_addr * src,
+                          const uint8_t *         hash)
+{
+        struct list_head * p;
+
+        pthread_once(&queries.once, queries_init);
+
+        pthread_mutex_lock(&queries.mtx);
+
+        list_for_each(p, &queries.pending.list) {
+                struct eth_query * q;
+
+                q = list_entry(p, struct eth_query, next);
+                if (q->replied || memcmp(q->hash, hash, POA_QUERY_HLEN) != 0)
+                        continue;
+
+                if (!eth_query_is_match(q, poa))
+                        continue;
+
+                memcpy(q->mac, src->eth.dst.mac, POA_MAC_SIZE);
+                memcpy(q->src_mac, src->eth.src.mac, POA_MAC_SIZE);
+                strcpy(q->dev, src->eth.src.dev);
+
+                q->r_ethertype = poa->local.eth.src.ethertype;
+
+                q->replied = true;
+        }
+
+        pthread_cond_broadcast(&queries.cond);
+
+        pthread_mutex_unlock(&queries.mtx);
+}
+
+/* Name query frames are handled in the transport. */
+static void eth_rx_mgmt(struct poa *            poa,
+                        const struct poa_addr * src,
+                        const uint8_t *         buf,
+                        size_t                  len)
+{
+        const struct poa_mgmt_msg * msg;
+        const uint8_t *             hash;
+
+        msg = (const struct poa_mgmt_msg *) buf;
+        if (len < sizeof(*msg)) {
+                poa_rx_mgmt(poa, src, buf, len);
+                return;
+        }
+
+        if (msg->code != POA_NAME_QUERY && msg->code != POA_NAME_REPLY) {
+                poa_rx_mgmt(poa, src, buf, len);
+                return;
+        }
+
+        if (ntoh16(msg->data_len) != POA_QUERY_HLEN)
+                return; /* malformed */
+
+        if (len < sizeof(*msg) + POA_QUERY_HLEN)
+                return; /* malformed */
+
+        hash = buf + sizeof(*msg);
+
+        if (msg->code == POA_NAME_QUERY) {
+                POA_STAT_BUMP(poa, qry_rx);
+                eth_query_arr(poa, src, hash);
+        } else {
+                POA_STAT_BUMP(poa, rep_rx);
+                eth_reply_arr(poa, src, hash);
+        }
+}
+
+static int eth_query_send(const uint8_t *         hash,
+                          const struct poa_addr * addr)
+{
+        uint8_t               buf[sizeof(struct poa_mgmt_msg)
+                                   + POA_QUERY_HLEN];
+        struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+        struct poa_addr       bcast;
+
+        bcast = *addr;
+
+        memset(bcast.eth.dst.mac, 0xff, POA_MAC_SIZE);
+
+        poa_mgmt_msg_ser(msg, POA_NAME_QUERY, 0, 0, qos_raw, 0, POA_QUERY_HLEN);
+
+        memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN);
+
+        return poa_bcast_mgmt(&bcast, buf, sizeof(buf));
+}
+
+/* Complete addr for dst by broadcast query; the poa_query op. */
+static int eth_query(const char *            dst,
+                     const struct timespec * timeo,
+                     struct poa_addr *       addr)
+{
+        struct timespec  dflt  = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO);
+        struct timespec  rintv = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO
+                                                  / (ETH_QUERY_RETRIES + 1));
+        struct eth_query q;
+        struct timespec  abstime;
+        struct timespec  now;
+        struct timespec  dl;
+        time_t           rintv_ns = (time_t) TS_TO_UINT64(rintv);
+        time_t           diff;
+        uint8_t          hash[POA_QUERY_HLEN];
+        int              n;
+        int              err = -ETIMEDOUT;
+
+        if (strlen(addr->eth.src.dev) > DEV_NAME_SIZE)
+                return -EINVAL;
+
+        addr->eth.src.ethertype = addr->eth.dst.ethertype;
+
+        /* The destination is set; a zero ethertype cannot be sent. */
+        if (!mac_is_zero(addr->eth.dst.mac))
+                return addr->eth.dst.ethertype < 0x0600 ? -EINVAL : 0;
+
+        str_hash(HASH_SHA3_256, hash, dst);
+
+        pthread_once(&queries.once, queries_init);
+
+        memset(&q, 0, sizeof(q));
+        memcpy(q.hash, hash, POA_QUERY_HLEN);
+
+        q.ethertype = addr->eth.dst.ethertype;
+
+        memcpy(q.c_mac, addr->eth.src.mac, POA_MAC_SIZE);
+        strcpy(q.c_dev, addr->eth.src.dev);
+
+        pthread_mutex_lock(&queries.mtx);
+        llist_add(&q.next, &queries.pending);
+        pthread_mutex_unlock(&queries.mtx);
+
+        pthread_cleanup_push(eth_query_cleanup, &q);
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+        ts_add(&now, timeo != NULL ? timeo : &dflt, &abstime);
+
+        while (ts_diff_ns(&now, &abstime) < 0) {
+                n = eth_query_send(hash, addr);
+                if (n < 0) {
+                        err = n;
+                        break;
+                }
+
+                if (n == 0) {
+                        err = -EPERM;
+                        break;
+                }
+
+                ts_add(&now, &rintv, &dl);
+
+                if (ts_diff_ns(&dl, &abstime) > 0)
+                        dl = abstime;
+
+                pthread_mutex_lock(&queries.mtx);
+
+                pthread_cleanup_push(__cleanup_mutex_unlock, &queries.mtx);
+
+                while (!q.replied) {
+                        if (pthread_cond_timedwait(&queries.cond, &queries.mtx,
+                                                   &dl) == ETIMEDOUT)
+                                break;
+                }
+
+                if (q.replied) {
+                        memcpy(addr->eth.dst.mac, q.mac, POA_MAC_SIZE);
+                        memcpy(addr->eth.src.mac, q.src_mac, POA_MAC_SIZE);
+                        strcpy(addr->eth.src.dev, q.dev);
+
+                        addr->eth.dst.ethertype = q.r_ethertype;
+                        addr->eth.src.ethertype = q.r_ethertype;
+
+                        err = 0;
+                }
+
+                pthread_cleanup_pop(true);
+
+                if (err == 0)
+                        break;
+
+                clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+                diff = ts_diff_ns(&now, &abstime);
+                if (diff > -rintv_ns) /* skip the runt attempt */
+                        break;
+        }
+
+        pthread_cleanup_pop(true);
+
+        return err;
+}
+
+static void eth_spec(const struct poa * poa,
+                     struct poa_spec *  spec)
+{
+        spec->type = poa->type;
+        spec->eth  = poa->local.eth.src;
+}
+
+static bool eth_has_id(const struct poa *      poa,
+                       const struct poa_spec * spec)
+{
+        if (strnlen(spec->eth.dev, sizeof(spec->eth.dev)) > DEV_NAME_SIZE)
+                return false;
+
+        if (strcmp(poa->local.eth.src.dev, spec->eth.dev) != 0)
+                return false;
+
+        return poa->local.eth.src.ethertype == spec->eth.ethertype;
+}
+
+/* Our end of the link; an unnamed one matches any. */
+static bool eth_is_src(const struct poa *     poa,
+                       const struct eth_poa * src)
+{
+        if (src->dev[0] != '\0' &&
+            strcmp(poa->local.eth.src.dev, src->dev) != 0)
+                return false;
+
+        if (mac_is_zero(src->mac))
+                return true;
+
+        return memcmp(poa->local.eth.src.mac, src->mac,
+                      POA_MAC_SIZE) == 0;
+}
+
+/* An ethertype of 0 in dst matches any, for broadcast queries. */
+static bool eth_match(const struct poa *      poa,
+                      const struct poa_addr * dst)
+{
+        uint16_t et = dst->eth.dst.ethertype;
+
+        if (et != 0 && et != poa->local.eth.src.ethertype)
+                return false;
+
+        return eth_is_src(poa, &dst->eth.src);
+}
+
+static bool eth_link_match(const struct poa * poa,
+                           int                id)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        return priv->ifindex == id;
+}
+
+#ifdef HAVE_RAW_SOCKETS
+
+/* MSG_DONTWAIT: the reader blocks on this socket. */
+static int eth_sendv(struct eth_priv *       priv,
+                     const uint8_t *         dst,
+                     uint32_t                eid,
+                     const uint8_t *         body,
+                     size_t                  len,
+                     bool                    block,
+                     const struct timespec * abstime)
+{
+        struct eth_hdr hdr;
+        struct msghdr  msg;
+        struct iovec   iov[2];
+        int            ret;
+
+        if (len > priv->mtu - POA_HDR_SIZE)
+                return -EMSGSIZE;
+
+        eth_hdr_ser(priv, &hdr, dst, eid, len);
+
+        iov[0].iov_base = &hdr;
+        iov[0].iov_len  = ETH_HDR_TOT_SIZE;
+        iov[1].iov_base = (void *) body;
+        iov[1].iov_len  = len;
+
+        memset(&msg, 0, sizeof(msg));
+
+        msg.msg_name    = &priv->device;
+        msg.msg_namelen = sizeof(priv->device);
+        msg.msg_iov     = iov;
+        msg.msg_iovlen  = len > 0 ? 2 : 1;
+        while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) {
+                if (errno != EAGAIN && errno != EWOULDBLOCK)
+                        return -EIO;
+
+                if (!block)
+                        return -EAGAIN;
+
+                ret = poa_wait_out(priv->s_fd, abstime);
+                if (ret < 0)
+                        return ret;
+        }
+
+        return 0;
+}
+
+static void * eth_reader(void * o)
+{
+        struct poa *      poa  = (struct poa *) o;
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+        struct timespec   down = TIMESPEC_INIT_MS(ETH_DOWN_TIMEO);
+        uint8_t *         buf;
+        size_t            bufsz;
+
+        bufsz = ETH_HDR_SIZE + priv->mtu;
+
+        buf = malloc(bufsz);
+        if (buf == NULL)
+                return (void *) -1;
+
+        pthread_cleanup_push(free, buf);
+
+        while (true) {
+                struct ssm_pk_buff * spb;
+                struct sockaddr_ll   from;
+                socklen_t            flen;
+                struct poa_addr      src;
+                ssize_t              n;
+                uint32_t             eid;
+                size_t               plen;
+                const uint8_t *      body;
+
+                flen = sizeof(from);
+
+                n = recvfrom(priv->s_fd, buf, bufsz, 0,
+                             (struct sockaddr *) &from, &flen);
+
+                if (n < 0) {
+                        if (errno == EINTR)
+                                continue;
+
+                        POA_STAT_BUMP(poa, rcv_fail);
+
+                        if (errno == ENETDOWN) {
+                                nanosleep(&down, NULL);
+                                continue;
+                        }
+
+                        log_err("Reader on %s stopped: %s.",
+                                poa->local.eth.src.dev,
+                                strerror(errno));
+                        break;
+                }
+
+                if (from.sll_pkttype == PACKET_OUTGOING)
+                        continue;
+
+                if (frame_parse(priv, buf, (size_t) n, &eid, &plen) < 0)
+                        continue;
+
+                body = buf + ETH_HDR_TOT_SIZE;
+
+                if (eid == POA_MGMT_EID) {
+                        frame_to_addr(priv, (struct eth_hdr *) buf, &src);
+                        eth_rx_mgmt(poa, &src, body, plen);
+                        continue;
+                }
+
+                if (poa_spb_reserve(&spb, plen) < 0) {
+                        POA_STAT_BUMP(poa, buf_fail);
+                        continue;
+                }
+
+                memcpy(ssm_pk_buff_head(spb), body, plen);
+
+                poa_rx_pkt(poa, eid, spb);
+        }
+
+        pthread_cleanup_pop(true);
+
+        return (void *) 0;
+}
+
+/*
+ * One netlink socket for the whole subsystem: RTMGRP_LINK delivers the
+ * events of every interface anyway, so a socket per PoA only added
+ * discards.
+ */
+int poa_monitor_open(void)
+{
+        struct sockaddr_nl sa;
+        int                fd;
+
+        memset(&sa, 0, sizeof(sa));
+
+        sa.nl_family = AF_NETLINK;
+        sa.nl_groups = RTMGRP_LINK;
+
+        fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE);
+        if (fd < 0)
+                return -1;
+
+        if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0) {
+                close(fd);
+                return -1;
+        }
+
+        return fd;
+}
+
+void poa_monitor_read(int fd)
+{
+        uint8_t           buf[4096];
+        struct nlmsghdr * h;
+        ssize_t           n;
+
+        n = recv(fd, buf, sizeof(buf), 0);
+        if (n < 0)
+                return;
+
+        for (h = (struct nlmsghdr *) buf;
+             NLMSG_OK(h, (unsigned int) n);
+             h = NLMSG_NEXT(h, n)) {
+                struct ifinfomsg * ifi;
+                unsigned int       usable;
+                bool               up;
+                size_t             cnt;
+
+                if (h->nlmsg_type == NLMSG_DONE)
+                        break;
+
+                if (h->nlmsg_type != RTM_NEWLINK)
+                        continue;
+
+                ifi = NLMSG_DATA(h);
+
+                usable = ifi->ifi_flags & (IFF_UP | IFF_RUNNING);
+                up     = usable == (IFF_UP | IFF_RUNNING);
+
+                cnt = poa_link_updown(ifi->ifi_index, up);
+                if (cnt > 0)
+                        log_info("Link %d %s, %zu flows.", ifi->ifi_index,
+                                 up ? "up" : "down", cnt);
+        }
+}
+
+static int eth_dev_info(const char * dev,
+                        int *        idx,
+                        uint8_t *    mac,
+                        uint32_t *   mtu)
+{
+        struct ifreq ifr;
+        int          fd;
+
+        if (strlen(dev) >= IFNAMSIZ)
+                return -EINVAL;
+
+        *idx = if_nametoindex(dev);
+        if (*idx == 0) {
+                log_err("Failed to find device %s.", dev);
+                return -ENODEV;
+        }
+
+        fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        if (fd < 0)
+                return -EIO;
+
+        memset(&ifr, 0, sizeof(ifr));
+        strcpy(ifr.ifr_name, dev);
+
+        if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) {
+                log_err("Failed to get hardware address of %s.", dev);
+                goto fail_ioctl;
+        }
+
+        memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE);
+
+        if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+                log_err("Failed to get MTU of %s.", dev);
+                goto fail_ioctl;
+        }
+
+        close(fd);
+
+        *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+        if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+                *mtu = POA_ETH_LO_MTU;
+
+        return 0;
+
+ fail_ioctl:
+        close(fd);
+        return -EIO;
+}
+
+/* SO_RCVBUFFORCE bypasses rmem_max; SO_RCVBUF is the fallback. */
+static void eth_set_rcvbuf(int fd,
+                           int rcvbuf)
+{
+        if (setsockopt(fd, SOL_SOCKET, SO_RCVBUFFORCE,
+                       &rcvbuf, sizeof(rcvbuf)) == 0)
+                return;
+
+        if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf)) < 0)
+                log_info("Failed to set SO_RCVBUF to %d.", rcvbuf);
+}
+
+/* The send buffer holds at least one frame, or sendmsg cannot fit a PDU. */
+static int eth_socket(struct eth_priv * priv,
+                      int               idx,
+                      const uint8_t *   mac)
+{
+        int sndbuf;
+        int rcvbuf;
+#ifdef POA_ETH_QDISC_BYPASS
+        int bypass = 1;
+#endif
+
+        memset(&priv->device, 0, sizeof(priv->device));
+
+        priv->device.sll_ifindex  = idx;
+        priv->device.sll_family   = AF_PACKET;
+        priv->device.sll_halen    = POA_MAC_SIZE;
+        priv->device.sll_protocol = htons(ETH_P_ALL);
+
+        memcpy(priv->device.sll_addr, mac, POA_MAC_SIZE);
+        memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+        priv->s_fd = socket(AF_PACKET, SOCK_RAW, priv->ethertype);
+        if (priv->s_fd < 0) {
+                log_err("Failed to create socket: %s.", strerror(errno));
+                return -1;
+        }
+
+#ifdef POA_ETH_QDISC_BYPASS
+        if (setsockopt(priv->s_fd, SOL_PACKET, PACKET_QDISC_BYPASS,
+                       &bypass, sizeof(bypass)) < 0)
+                log_info("Qdisc bypass not supported.");
+#endif
+
+        sndbuf = POA_ETH_SNDBUF;
+        if (sndbuf > 0) {
+                sndbuf = MAX(sndbuf, (int) (ETH_HDR_SIZE + priv->mtu));
+
+                if (setsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF,
+                               &sndbuf, sizeof(sndbuf)) < 0)
+                        log_info("Failed to set SO_SNDBUF to %d.", sndbuf);
+        }
+
+        rcvbuf = POA_ETH_RCVBUF;
+        if (rcvbuf > 0)
+                eth_set_rcvbuf(priv->s_fd, rcvbuf);
+
+        if (bind(priv->s_fd, (struct sockaddr *) &priv->device,
+                 sizeof(priv->device)) < 0) {
+                log_err("Failed to bind socket to %d.", idx);
+                close(priv->s_fd);
+                return -1;
+        }
+
+        return 0;
+}
+
+/*
+ * Pull qlen and backlog from the nested TCA_STATS2. The top-level
+ * TCA_STATS shares TCA_STATS_QUEUE's id but carries a wholly
+ * different struct, so descend first; never match on id alone.
+ */
+static void eth_qdisc_parse(struct rtattr * rta,
+                            size_t          rlen,
+                            size_t *        byt,
+                            size_t *        pkt)
+{
+        struct gnet_stats_queue q;
+        struct rtattr *         in;
+        size_t                  ilen;
+
+        for (; RTA_OK(rta, rlen); rta = RTA_NEXT(rta, rlen)) {
+                if (rta->rta_type != TCA_STATS2)
+                        continue;
+
+                in   = (struct rtattr *) RTA_DATA(rta);
+                ilen = RTA_PAYLOAD(rta);
+                for (; RTA_OK(in, ilen); in = RTA_NEXT(in, ilen)) {
+                        if (in->rta_type != TCA_STATS_QUEUE)
+                                continue;
+
+                        if (RTA_PAYLOAD(in) < sizeof(q))
+                                continue;
+
+                        memcpy(&q, RTA_DATA(in), sizeof(q));
+
+                        *byt = q.backlog;
+                        *pkt = q.qlen;
+                }
+        }
+}
+
+/*
+ * Egress backlog of the device's root qdisc, in bytes and packets,
+ * as the kernel queues them. The caller owns the netlink descriptor
+ * and serialises the query. An early end of dump reports failure.
+ */
+static int eth_qdisc_backlog(int      fd,
+                             int      ifindex,
+                             size_t * byt,
+                             size_t * pkt)
+{
+        struct {
+                struct nlmsghdr nh;
+                struct tcmsg    tc;
+        }                  req;
+        struct nlmsghdr *  nh;
+        struct rtattr *    rta;
+        struct tcmsg *     tc;
+        char               buf[16384];
+        ssize_t            len;
+        int                ret = -1;
+
+        if (fd < 0)
+                goto fail;
+
+        memset(&req, 0, sizeof(req));
+
+        req.nh.nlmsg_len   = NLMSG_LENGTH(sizeof(req.tc));
+        req.nh.nlmsg_type  = RTM_GETQDISC;
+        req.nh.nlmsg_flags = NLM_F_REQUEST | NLM_F_DUMP;
+        req.nh.nlmsg_seq   = 1;
+        req.tc.tcm_family  = AF_UNSPEC;
+        req.tc.tcm_ifindex = ifindex;
+        if (send(fd, &req, req.nh.nlmsg_len, 0) < 0)
+                goto fail;
+
+        *byt = 0;
+        *pkt = 0;
+
+        while ((len = recv(fd, buf, sizeof(buf), 0)) > 0) {
+                nh = (struct nlmsghdr *) buf;
+                for (; NLMSG_OK(nh, len); nh = NLMSG_NEXT(nh, len)) {
+                        if (nh->nlmsg_type == NLMSG_DONE)
+                                goto done;
+
+                        if (nh->nlmsg_type == NLMSG_ERROR)
+                                goto fail;
+
+                        if (nh->nlmsg_type != RTM_NEWQDISC)
+                                continue;
+
+                        tc = (struct tcmsg *) NLMSG_DATA(nh);
+                        if (tc->tcm_ifindex != ifindex)
+                                continue;
+
+                        if (tc->tcm_parent != TC_H_ROOT)
+                                continue;
+
+                        rta = (struct rtattr *)
+                                ((char *) tc + NLMSG_ALIGN(sizeof(*tc)));
+
+                        eth_qdisc_parse(rta, nh->nlmsg_len
+                                        - NLMSG_LENGTH(sizeof(*tc)),
+                                        byt, pkt);
+                }
+        }
+
+        goto fail; /* early end: a zero would read as empty */
+ done:
+        ret = 0;
+ fail:
+        return ret;
+}
+
+/*
+ * A netlink descriptor for the qdisc query. SO_RCVTIMEO bounds the
+ * dump: this is read from the send path, and a reply that never
+ * arrives must not park a sender thread.
+ */
+static int eth_qdisc_open(void)
+{
+        struct sockaddr_nl sa;
+        struct timeval     tv = TIMEVAL_INIT_MS(POA_QDISC_TIMEO);
+        int                fd;
+
+        fd = socket(AF_NETLINK, SOCK_RAW | SOCK_CLOEXEC, NETLINK_ROUTE);
+        if (fd < 0)
+                return -1;
+
+        memset(&sa, 0, sizeof(sa));
+
+        sa.nl_family = AF_NETLINK;
+        if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0)
+                goto fail;
+
+        if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0)
+                goto fail;
+
+        return fd;
+ fail:
+        close(fd);
+
+        return -1;
+}
+
+/*
+ * Queue depth in packets, straight from the qdisc. Gated like the
+ * transport's own depth and skipped when another sender is already
+ * asking, so the send path never waits on netlink.
+ */
+static int eth_qpkts(struct poa * poa,
+                     size_t *     pkts,
+                     size_t *     byts)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+        struct timespec   now;
+        uint64_t          ns;
+        size_t            byt;
+        size_t            pkt;
+
+        if (priv->nl_fd < 0)
+                return -1;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+        ns = TS_TO_UINT64(now);
+        if (ns - LOAD_RELAXED(&priv->nl_time) < POA_QLEN_GATE)
+                goto cached;
+
+        if (__atomic_exchange_n(&priv->nl_busy, 1, __ATOMIC_ACQUIRE) != 0)
+                goto cached;
+
+        if (eth_qdisc_backlog(priv->nl_fd, priv->ifindex, &byt, &pkt) == 0) {
+                STORE_RELAXED(&priv->nl_pkt, pkt);
+                STORE_RELAXED(&priv->nl_byt, byt);
+                STORE_RELAXED(&priv->nl_time, ns);
+        }
+
+        __atomic_store_n(&priv->nl_busy, 0, __ATOMIC_RELEASE);
+ cached:
+        if (LOAD_RELAXED(&priv->nl_time) == 0) /* nothing measured yet */
+                return -1;
+
+        *pkts = LOAD_RELAXED(&priv->nl_pkt);
+        *byts = LOAD_RELAXED(&priv->nl_byt);
+
+        return 0;
+}
+
+static int eth_attach(struct poa *            poa,
+                      const struct poa_spec * spec)
+{
+        const struct eth_poa * c = &spec->eth;
+        struct eth_priv *      priv;
+        uint8_t                mac[POA_MAC_SIZE];
+        uint32_t               mtu;
+        int                    idx;
+        int                    err;
+
+        if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+                return -EINVAL;
+
+        if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+                log_err("Invalid Ethertype 0x%x.", c->ethertype);
+                return -EINVAL;
+        }
+
+        priv = malloc(sizeof(*priv));
+        if (priv == NULL)
+                return -ENOMEM;
+
+        memset(priv, 0, sizeof(*priv));
+
+        priv->poa   = poa;
+        priv->nl_fd = eth_qdisc_open(); /* optional; socket fallback */
+
+        priv->ethertype = htons(c->ethertype);
+
+        err = eth_dev_info(c->dev, &idx, mac, &mtu);
+        if (err < 0)
+                goto fail_conf;
+
+        priv->mtu = mtu;
+        if (eth_socket(priv, idx, mac) < 0)
+                goto fail_conf;
+
+        poa->priv                    = priv;
+        poa->local.type              = poa->type;
+        poa->local.eth.src.ethertype = c->ethertype;
+        priv->ifindex                = idx;
+
+        memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+        strcpy(poa->local.eth.src.dev, c->dev);
+
+        return 0;
+
+ fail_conf:
+        if (priv->nl_fd >= 0)
+                close(priv->nl_fd);
+
+        free(priv);
+
+        return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        if (priv == NULL)
+                return;
+
+        close(priv->s_fd);
+
+        if (priv->nl_fd >= 0)
+                close(priv->nl_fd);
+
+        free(priv);
+
+        poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa *            poa,
+                        const struct poa_addr * dst)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        (void) dst;
+
+        return priv->mtu - POA_HDR_SIZE;
+}
+
+/* All flows on the PoA share the socket, so this is aggregate. */
+static size_t eth_qlen(struct poa * poa)
+{
+#ifdef SIOCOUTQ
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+        int               qlen;
+
+        qlen = 0;
+        if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0)
+                return 0;
+
+        return (size_t) qlen;
+#else
+        (void) poa;
+
+        return 0;
+#endif
+}
+
+/*
+ * PACKET_STATISTICS zeroes the kernel counters on read, so totals
+ * accumulate here; relaxed atomics allow concurrent RIB reads.
+ * The qdisc depth rides the sender's guarded path: one dump at a time.
+ */
+static int eth_rib(struct poa * poa,
+                   char *       buf,
+                   size_t       len)
+{
+        struct eth_priv *    priv = (struct eth_priv *) poa->priv;
+        struct tpacket_stats ts;
+        socklen_t            optlen;
+        size_t               sndbuf = 0;
+        size_t               rcvbuf = 0;
+        size_t               qd_byt;
+        size_t               qd_pkt;
+        int                  val;
+        int                  size;
+
+        optlen = sizeof(val);
+        if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0)
+                sndbuf = (size_t) val;
+
+        optlen = sizeof(val);
+        if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0)
+                rcvbuf = (size_t) val;
+
+        optlen = sizeof(ts);
+
+        if (getsockopt(priv->s_fd, SOL_PACKET, PACKET_STATISTICS,
+                       &ts, &optlen) == 0) {
+                FETCH_ADD_RELAXED(&priv->kern_rcv, ts.tp_packets);
+                FETCH_ADD_RELAXED(&priv->kern_drp, ts.tp_drops);
+        }
+
+        if (eth_qpkts(poa, &qd_pkt, &qd_byt) < 0) {
+                qd_byt = 0;
+                qd_pkt = 0;
+        }
+
+        size = snprintf(buf, len,
+                        "Socket sndbuf (bytes):     %zu\n"
+                        "Socket rcvbuf (bytes):     %zu\n"
+                        "Kernel packets received:   %zu\n"
+                        "Kernel packets dropped:    %zu\n"
+                        "Qdisc backlog (bytes):     %zu\n"
+                        "Qdisc backlog (packets):   %zu\n",
+                        sndbuf, rcvbuf,
+                        LOAD_RELAXED(&priv->kern_rcv),
+                        LOAD_RELAXED(&priv->kern_drp),
+                        qd_byt, qd_pkt);
+        if (size < 0 || (size_t) size >= len)
+                return -1;
+
+        return size;
+}
+
+#elif defined(HAVE_BPF)
+
+/*
+ * BSD and macOS reach the link layer through a cloned /dev/bpf
+ * device, bound to an interface with BIOCSETIF. One device per PoA.
+ */
+
+static int eth_sendv(struct eth_priv *       priv,
+                     const uint8_t *         dst,
+                     uint32_t                eid,
+                     const uint8_t *         body,
+                     size_t                  len,
+                     bool                    block,
+                     const struct timespec * abstime)
+{
+        struct eth_hdr hdr;
+        struct iovec   iov[2];
+        int            n;
+        int            ret;
+
+        if (len > priv->mtu - POA_HDR_SIZE)
+                return -EMSGSIZE;
+
+        eth_hdr_ser(priv, &hdr, dst, eid, len);
+
+        iov[0].iov_base = &hdr;
+        iov[0].iov_len  = ETH_HDR_TOT_SIZE;
+        iov[1].iov_base = (void *) body;
+        iov[1].iov_len  = len;
+        n               = len > 0 ? 2 : 1;
+        while (writev(priv->s_fd, iov, n) < 0) {
+                if (errno != EAGAIN && errno != EWOULDBLOCK)
+                        return -EIO;
+
+                if (!block)
+                        return -EAGAIN;
+
+                ret = poa_wait_out(priv->s_fd, abstime);
+                if (ret < 0)
+                        return ret;
+        }
+
+        return 0;
+}
+
+/* One read yields a batch of BPF_WORDALIGN'ed frames; walk all of them. */
+static void * eth_reader(void * o)
+{
+        struct poa *      poa  = (struct poa *) o;
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+        uint8_t *         buf;
+
+        buf = malloc(priv->blen);
+        if (buf == NULL)
+                return (void *) -1;
+
+        pthread_cleanup_push(free, buf);
+
+        while (true) {
+                uint8_t * p;
+                uint8_t * end;
+                ssize_t   n;
+
+                n = read(priv->s_fd, buf, priv->blen);
+                if (n < 0) {
+                        if (errno == EINTR)
+                                continue;
+
+                        POA_STAT_BUMP(poa, rcv_fail);
+                        break;
+                }
+
+                p   = buf;
+                end = buf + n;
+                while (p + sizeof(struct bpf_hdr) <= end) {
+                        struct bpf_hdr *     bh = (struct bpf_hdr *) p;
+                        struct ssm_pk_buff * spb;
+                        struct poa_addr      src;
+                        const uint8_t *      frame;
+                        const uint8_t *      body;
+                        uint32_t             eid;
+                        size_t               plen;
+
+                        frame = p + bh->bh_hdrlen;
+                        if (frame + bh->bh_caplen > end)
+                                break;
+
+                        if (frame_parse(priv, frame, bh->bh_caplen,
+                                        &eid, &plen) < 0)
+                                goto next;
+
+                        body = frame + ETH_HDR_TOT_SIZE;
+
+                        if (eid == POA_MGMT_EID) {
+                                frame_to_addr(priv,
+                                              (const struct eth_hdr *) frame,
+                                              &src);
+                                eth_rx_mgmt(poa, &src, body, plen);
+                                goto next;
+                        }
+
+                        if (poa_spb_reserve(&spb, plen) < 0) {
+                                POA_STAT_BUMP(poa, buf_fail);
+                                goto next;
+                        }
+
+                        memcpy(ssm_pk_buff_head(spb), body, plen);
+
+                        poa_rx_pkt(poa, eid, spb);
+ next:
+                        p += BPF_WORDALIGN(bh->bh_hdrlen + bh->bh_caplen);
+                }
+        }
+
+        pthread_cleanup_pop(true);
+
+        return (void *) 0;
+}
+
+static int eth_dev_info(const char * dev,
+                        uint8_t *    mac,
+                        uint32_t *   mtu)
+{
+        struct ifaddrs * ifas;
+        struct ifaddrs * ifa;
+        struct ifreq     ifr;
+        int              fd;
+        int              found = 0;
+
+        if (getifaddrs(&ifas) < 0) {
+                log_err("Failed to list interfaces.");
+                return -1;
+        }
+
+        for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) {
+                struct sockaddr_dl * dl;
+
+                if (ifa->ifa_addr == NULL)
+                        continue;
+
+                if (ifa->ifa_addr->sa_family != AF_LINK)
+                        continue;
+
+                if (strcmp(ifa->ifa_name, dev) != 0)
+                        continue;
+
+                dl = (struct sockaddr_dl *) ifa->ifa_addr;
+                if (dl->sdl_alen != POA_MAC_SIZE)
+                        continue;
+
+                memcpy(mac, LLADDR(dl), POA_MAC_SIZE);
+
+                found = 1;
+                break;
+        }
+
+        freeifaddrs(ifas);
+
+        if (!found) {
+                log_err("No such device: %s.", dev);
+                return -1;
+        }
+
+        fd = socket(AF_INET, SOCK_DGRAM, 0);
+        if (fd < 0)
+                return -1;
+
+        memset(&ifr, 0, sizeof(ifr));
+
+        strcpy(ifr.ifr_name, dev);
+
+        if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+                log_err("Failed to get MTU of %s.", dev);
+                close(fd);
+                return -1;
+        }
+
+        close(fd);
+
+        *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+        if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+                *mtu = POA_ETH_LO_MTU;
+
+        return 0;
+}
+
+static int eth_bpf_open(void)
+{
+        char   dev[32];
+        size_t i;
+
+        for (i = 0; i < BPF_DEV_MAX; ++i) {
+                int fd;
+
+                sprintf(dev, "/dev/bpf%zu", i);
+
+                fd = open(dev, O_RDWR);
+                if (fd >= 0)
+                        return fd;
+        }
+
+        return -1;
+}
+
+/*
+ * BIOCSHDRCMPLT: we fill in the source address ourselves.
+ * BIOCSSEESENT:  our own egress must not come back at us.
+ * BIOCIMMEDIATE: deliver on arrival, do not wait for a full buffer.
+ */
+static int eth_bpf_setup(struct eth_priv * priv,
+                         const char *      dev)
+{
+        struct ifreq ifr;
+        int          enable  = 1;
+        int          disable = 0;
+        int          blen    = 0;
+
+        memset(&ifr, 0, sizeof(ifr));
+
+        strcpy(ifr.ifr_name, dev);
+
+        if (ioctl(priv->s_fd, BIOCSETIF, &ifr) < 0) {
+                log_err("Failed to bind bpf device to %s.", dev);
+                return -1;
+        }
+
+        if (ioctl(priv->s_fd, BIOCGBLEN, &blen) < 0 || blen <= 0) {
+                log_err("Failed to get the bpf buffer length.");
+                return -1;
+        }
+
+        priv->blen = (size_t) blen;
+        if (ioctl(priv->s_fd, BIOCSHDRCMPLT, &enable) < 0) {
+                log_err("Failed to set BIOCSHDRCMPLT.");
+                return -1;
+        }
+
+        if (ioctl(priv->s_fd, BIOCSSEESENT, &disable) < 0) {
+                log_err("Failed to set BIOCSSEESENT.");
+                return -1;
+        }
+
+        if (ioctl(priv->s_fd, BIOCIMMEDIATE, &enable) < 0) {
+                log_err("Failed to set BIOCIMMEDIATE.");
+                return -1;
+        }
+
+        return 0;
+}
+
+static int eth_attach(struct poa *            poa,
+                      const struct poa_spec * spec)
+{
+        const struct eth_poa * c = &spec->eth;
+        struct eth_priv *      priv;
+        uint8_t                mac[POA_MAC_SIZE];
+        uint32_t               mtu;
+
+        if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+                return -EINVAL;
+
+        if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+                log_err("Invalid Ethertype 0x%x.", c->ethertype);
+                return -EINVAL;
+        }
+
+        priv = malloc(sizeof(*priv));
+        if (priv == NULL)
+                return -ENOMEM;
+
+        memset(priv, 0, sizeof(*priv));
+
+        priv->poa  = poa;
+        priv->s_fd = -1;
+
+        priv->ethertype = htons(c->ethertype);
+
+        if (eth_dev_info(c->dev, mac, &mtu) < 0)
+                goto fail_conf;
+
+        priv->mtu = mtu;
+
+        memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+        priv->s_fd = eth_bpf_open();
+        if (priv->s_fd < 0) {
+                log_err("Failed to open a bpf device.");
+                goto fail_conf;
+        }
+
+        if (eth_bpf_setup(priv, c->dev) < 0)
+                goto fail_device;
+
+        poa->priv                    = priv;
+        poa->local.type              = poa->type;
+        poa->local.eth.src.ethertype = c->ethertype;
+
+        memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+        strcpy(poa->local.eth.src.dev, c->dev);
+
+        log_info("Using Berkeley Packet Filter on %s.", c->dev);
+
+        return 0;
+
+ fail_device:
+        close(priv->s_fd);
+ fail_conf:
+        free(priv);
+
+        return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        if (priv == NULL)
+                return;
+
+        close(priv->s_fd);
+
+        free(priv);
+
+        poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa *            poa,
+                        const struct poa_addr * dst)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        (void) dst;
+
+        return priv->mtu - POA_HDR_SIZE;
+}
+
+/* A bpf device has no send queue to report; mb-ECN cannot mark here. */
+static size_t eth_qlen(struct poa * poa)
+{
+        (void) poa;
+
+        return 0;
+}
+
+/* The bpf device buffer is all the receive queue there is. */
+static int eth_rib(struct poa * poa,
+                   char *       buf,
+                   size_t       len)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+        struct bpf_stat   bs;
+        int               size;
+
+        if (ioctl(priv->s_fd, BIOCGSTATS, &bs) == 0) {
+                FETCH_ADD_RELAXED(&priv->kern_rcv, bs.bs_recv);
+                FETCH_ADD_RELAXED(&priv->kern_drp, bs.bs_drop);
+        }
+
+        size = snprintf(buf, len,
+                        "Socket rcvbuf (bytes):     %zu\n"
+                        "Kernel packets received:   %zu\n"
+                        "Kernel packets dropped:    %zu\n",
+                        priv->blen,
+                        LOAD_RELAXED(&priv->kern_rcv),
+                        LOAD_RELAXED(&priv->kern_drp));
+        if (size < 0 || (size_t) size >= len)
+                return -1;
+
+        return size;
+}
+
+#elif defined(HAVE_NETMAP)
+
+/*
+ * netmap gives one PoA the whole port: reads copy out of the NIC
+ * ring, writes inject into it.
+ */
+
+/* nm_inject takes one contiguous frame, so the header is copied in. */
+static int eth_sendv(struct eth_priv *       priv,
+                     const uint8_t *         dst,
+                     uint32_t                eid,
+                     const uint8_t *         body,
+                     size_t                  len,
+                     bool                    block,
+                     const struct timespec * abstime)
+{
+        uint8_t * frame;
+        size_t    flen;
+        int       ret;
+
+        if (len > priv->mtu - POA_HDR_SIZE)
+                return -EMSGSIZE;
+
+        flen  = ETH_HDR_TOT_SIZE + len;
+
+        frame = malloc(flen);
+        if (frame == NULL)
+                return -ENOMEM;
+
+        eth_hdr_ser(priv, (struct eth_hdr *) frame, dst, eid, len);
+
+        if (len > 0)
+                memcpy(frame + ETH_HDR_TOT_SIZE, body, len);
+
+        if (block)
+                ret = poa_wait_out(priv->poll_out.fd, abstime);
+        else
+                ret = poll(&priv->poll_out, 1, 0) > 0 ? 0 : -EAGAIN;
+
+        if (ret < 0)
+                goto fail;
+
+        ret = nm_inject(priv->nmd, frame, flen) == (int) flen ? 0 : -EIO;
+ fail:
+        free(frame);
+
+        return ret;
+}
+
+/* A slot stays owned by the ring, so each frame is copied out. */
+static void * eth_reader(void * o)
+{
+        struct poa *      poa  = (struct poa *) o;
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        while (true) {
+                struct ssm_pk_buff * spb;
+                struct poa_addr      src;
+                struct nm_pkthdr     hdr;
+                const uint8_t *      frame;
+                const uint8_t *      body;
+                uint32_t             eid;
+                size_t               plen;
+
+                if (poll(&priv->poll_in, 1, -1) < 0) {
+                        if (errno == EINTR)
+                                continue;
+
+                        POA_STAT_BUMP(poa, rcv_fail);
+                        break;
+                }
+
+                if (priv->poll_in.revents == 0)
+                        continue;
+
+                frame = nm_nextpkt(priv->nmd, &hdr);
+                if (frame == NULL)
+                        continue;
+
+                if (frame_parse(priv, frame, hdr.len, &eid, &plen) < 0)
+                        continue;
+
+                body = frame + ETH_HDR_TOT_SIZE;
+
+                if (eid == POA_MGMT_EID) {
+                        frame_to_addr(priv, (const struct eth_hdr *) frame,
+                                      &src);
+                        eth_rx_mgmt(poa, &src, body, plen);
+                        continue;
+                }
+
+                if (poa_spb_reserve(&spb, plen) < 0) {
+                        POA_STAT_BUMP(poa, buf_fail);
+                        continue;
+                }
+
+                memcpy(ssm_pk_buff_head(spb), body, plen);
+
+                poa_rx_pkt(poa, eid, spb);
+        }
+
+        return (void *) 0;
+}
+
+static int eth_dev_info(const char * dev,
+                        uint8_t *    mac,
+                        uint32_t *   mtu)
+{
+        struct ifreq     ifr;
+#ifndef __linux__
+        struct ifaddrs * ifas;
+        struct ifaddrs * ifa;
+        int              found = 0;
+#endif
+        int              fd;
+
+        if (strlen(dev) >= IFNAMSIZ)
+                return -EINVAL;
+
+        fd = socket(AF_INET, SOCK_DGRAM, 0);
+        if (fd < 0)
+                return -EIO;
+
+        memset(&ifr, 0, sizeof(ifr));
+
+        strcpy(ifr.ifr_name, dev);
+
+#ifdef __linux__
+        if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) {
+                log_err("Failed to get hardware address of %s.", dev);
+                goto fail_ioctl;
+        }
+
+        memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE);
+#else
+        if (getifaddrs(&ifas) < 0)
+                goto fail_ioctl;
+
+        for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) {
+                struct sockaddr_dl * dl;
+
+                if (ifa->ifa_addr == NULL)
+                        continue;
+
+                if (ifa->ifa_addr->sa_family != AF_LINK)
+                        continue;
+
+                if (strcmp(ifa->ifa_name, dev) != 0)
+                        continue;
+
+                dl = (struct sockaddr_dl *) ifa->ifa_addr;
+                if (dl->sdl_alen != POA_MAC_SIZE)
+                        continue;
+
+                memcpy(mac, LLADDR(dl), POA_MAC_SIZE);
+
+                found = 1;
+                break;
+        }
+
+        freeifaddrs(ifas);
+
+        if (!found) {
+                log_err("No hardware address for %s.", dev);
+                goto fail_ioctl;
+        }
+#endif
+        if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) {
+                log_err("Failed to get MTU of %s.", dev);
+                goto fail_ioctl;
+        }
+
+        close(fd);
+
+        *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu);
+        if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU)
+                *mtu = POA_ETH_LO_MTU;
+
+        return 0;
+
+ fail_ioctl:
+        close(fd);
+
+        return -EIO;
+}
+
+static int eth_nm_open(struct eth_priv * priv,
+                       const char *      dev)
+{
+        char ifn[IFNAMSIZ + sizeof("netmap:")];
+
+        strcpy(ifn, "netmap:");
+        strcat(ifn, dev);
+
+        priv->nmd = nm_open(ifn, NULL, 0, NULL);
+        if (priv->nmd == NULL) {
+                log_err("Failed to open netmap device for %s.", dev);
+                return -1;
+        }
+
+        memset(&priv->poll_in, 0, sizeof(priv->poll_in));
+        memset(&priv->poll_out, 0, sizeof(priv->poll_out));
+
+        priv->poll_in.fd      = NETMAP_FD(priv->nmd);
+        priv->poll_in.events  = POLLIN;
+        priv->poll_out.fd     = NETMAP_FD(priv->nmd);
+        priv->poll_out.events = POLLOUT;
+
+        return 0;
+}
+
+static int eth_attach(struct poa *            poa,
+                      const struct poa_spec * spec)
+{
+        const struct eth_poa * c = &spec->eth;
+        struct eth_priv *      priv;
+        uint8_t                mac[POA_MAC_SIZE];
+        uint32_t               mtu;
+
+        if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE)
+                return -EINVAL;
+
+        if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) {
+                log_err("Invalid Ethertype 0x%x.", c->ethertype);
+                return -EINVAL;
+        }
+
+        priv = malloc(sizeof(*priv));
+        if (priv == NULL)
+                return -ENOMEM;
+
+        memset(priv, 0, sizeof(*priv));
+
+        priv->poa = poa;
+
+        priv->ethertype = htons(c->ethertype);
+
+        if (eth_dev_info(c->dev, mac, &mtu) < 0)
+                goto fail_conf;
+
+        priv->mtu = mtu;
+
+        memcpy(priv->hw_addr, mac, POA_MAC_SIZE);
+
+        if (eth_nm_open(priv, c->dev) < 0)
+                goto fail_conf;
+
+        poa->priv                    = priv;
+        poa->local.type              = poa->type;
+        poa->local.eth.src.ethertype = c->ethertype;
+
+        memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE);
+        strcpy(poa->local.eth.src.dev, c->dev);
+
+        log_info("Using netmap on %s.", c->dev);
+
+        return 0;
+
+ fail_conf:
+        free(priv);
+
+        return -EIO;
+}
+
+static void eth_detach(struct poa * poa)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        if (priv == NULL)
+                return;
+
+        nm_close(priv->nmd);
+
+        free(priv);
+
+        poa->priv = NULL;
+}
+
+static uint32_t eth_mtu(struct poa *            poa,
+                        const struct poa_addr * dst)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        (void) dst;
+
+        return priv->mtu - POA_HDR_SIZE;
+}
+
+/* The ring is drained by the NIC; there is no queue to report. */
+static size_t eth_qlen(struct poa * poa)
+{
+        (void) poa;
+
+        return 0;
+}
+
+#endif /* HAVE_RAW_SOCKETS */
+
+#ifndef HAVE_RAW_SOCKETS
+
+/* Only netlink reports link events; no other backend has a monitor. */
+int poa_monitor_open(void)
+{
+        return -1;
+}
+
+void poa_monitor_read(int fd)
+{
+        (void) fd;
+}
+
+#endif
+
+/* One reader per socket, so a flow cannot be reordered on receive. */
+static int eth_start(struct poa * poa)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        if (pthread_create(&priv->reader, NULL, eth_reader, poa) != 0)
+                return -1;
+
+        priv->running = true;
+
+        return 0;
+}
+
+static void eth_stop(struct poa * poa)
+{
+        struct eth_priv * priv = (struct eth_priv *) poa->priv;
+
+        if (!priv->running)
+                return;
+
+        pthread_cancel(priv->reader);
+        pthread_join(priv->reader, NULL);
+
+        priv->running = false;
+}
+
+static int eth_send(struct poa *            poa,
+                    const struct poa_addr * dst,
+                    uint32_t                eid,
+                    struct ssm_pk_buff *    spb,
+                    bool                    block,
+                    const struct timespec * abstime)
+{
+        return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac, eid,
+                         ssm_pk_buff_head(spb), ssm_pk_buff_len(spb),
+                         block, abstime);
+}
+
+static int eth_send_mgmt(struct poa *            poa,
+                         const struct poa_addr * dst,
+                         const uint8_t *         buf,
+                         size_t                  len)
+{
+        struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO);
+        struct timespec abstime;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+        ts_add(&abstime, &timeo, &abstime);
+
+        return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac,
+                         POA_MGMT_EID, buf, len, true, &abstime);
+}
+
+const struct poa_ops eth_poa_ops = {
+        .poa_attach     = eth_attach,
+        .poa_detach     = eth_detach,
+        .poa_start      = eth_start,
+        .poa_stop       = eth_stop,
+        .poa_send       = eth_send,
+        .poa_send_mgmt  = eth_send_mgmt,
+        .poa_query      = eth_query,
+        .poa_mtu        = eth_mtu,
+        .poa_qlen       = eth_qlen,
+#ifdef HAVE_RAW_SOCKETS
+        .poa_qpkts      = eth_qpkts,
+#endif
+#ifndef HAVE_NETMAP
+        .poa_rib        = eth_rib,
+#endif
+        .poa_spec       = eth_spec,
+        .poa_has_id     = eth_has_id,
+        .poa_match      = eth_match,
+        .poa_link_match = eth_link_match,
+        .mpl            = POA_ETH_MPL
+};
diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c
new file mode 100644
index 00000000..3ad17c4f
--- /dev/null
+++ b/src/lib/poa/poa.c
@@ -0,0 +1,2515 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - transport independent core
+ *
+ * Included by dev.c; uses dev.c statics (proc, flow_init, ...).
+ *
+ *    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/.
+ */
+
+#define POA_MAX_EIDS     PROC_MAX_FLOWS
+#define POA_ALLOC_TIMEO  10000 /* ms, overall FLOW_REQ deadline  */
+#define POA_RETRY_TIMEO  300   /* ms, FLOW_REQ retransmit period */
+/* Must fit a certificate chain: post-quantum ones are large. */
+#define POA_MGMT_BUFSZ   POA_MGMT_FRAME_SIZE
+#define POA_MGMT_QMAX    64   /* queued management frames per PoA */
+#define POA_PEND_TIMEO   10   /* s, reap a request that never completes */
+#define POA_SWEEP_TIMEO  1000 /* ms, sweep interval */
+#define POA_SWEEP_MAX    16   /* requests reaped per sweep */
+#define POA_DEFER_MAX    64   /* replies waiting for their flow id */
+
+/* EWMA over 8 samples. */
+#define POA_AVG_SHIFT    3
+/* Queue cost is sampled every 64th packet: qlen is a syscall. */
+#define POA_COST_MASK    63
+/* Reuse a qlen read for this long; the mark moves on doublings. */
+
+#define POA_RIB          "poa"
+/* Fits the RIB labels below with 20-digit counters. */
+#define POA_RIB_STRLEN   2048
+
+enum poa_flow_state {
+        POA_FLOW_NULL = 0,
+        POA_FLOW_PENDING,
+        POA_FLOW_ALLOCATED,
+        POA_FLOW_DEAD
+};
+
+enum poa_state {
+        POA_NULL = 0,
+        POA_INIT,        /* poa_init(); this process may attach   */
+        POA_RUNNING,     /* the threads are up                    */
+        POA_OPERATIONAL  /* bootstrapped or enrolled in a layer   */
+};
+
+struct poa_flow {
+        struct poa *        poa;
+
+        int                 flow_id;
+        uint32_t            eid;
+        uint32_t            r_eid;
+        struct poa_addr     r_addr;
+
+        enum poa_flow_state state;
+
+        struct ssm_rbuff *  rx_rb;
+
+        struct list_head    pend;     /* on poa->pending while unanswered */
+        struct timespec     t0;
+
+        /* Our answer, kept to re-send when a request is repeated. */
+        bool                answered;
+        bool                answer_sent;
+
+        /* Handed to a caller that will attach it; not the sweeper's. */
+        bool                claimed;
+        int                 answer;
+        buffer_t            answer_data;
+
+        /* Handshake rendezvous with the reader thread. */
+        pthread_mutex_t     mtx;
+        pthread_cond_t      cond;
+        bool                replied;
+        bool                pending;
+        int                 response;
+        buffer_t            resp_data;
+};
+
+/*
+ * An answer can be ready before the flow it answers has an id, and
+ * the caller must not be kept waiting for one: it answers to the IRMd,
+ * which gives up long before we would.
+ */
+struct poa_deferred {
+        struct list_head next;
+        struct timespec  t0;
+        int              flow_id;
+        int              response;
+        buffer_t         data;
+};
+
+struct poa_mgmt_frame {
+        struct list_head next;
+        struct poa *     poa;
+        struct poa_addr  src;
+        size_t           len;
+        uint8_t          buf[POA_MGMT_BUFSZ];
+};
+
+/*
+ * A detach is performed by the management thread, so that it cannot
+ * run alongside the accept side. The caller waits for the result.
+ */
+struct poa_detach_req {
+        struct list_head next;
+        struct poa_spec  spec;
+        int              result;
+        bool             done;
+};
+
+static struct {
+        struct list_head  list;
+
+        struct poa_flow * id_to_pf[SYS_MAX_FLOWS];
+
+        struct llist      deferred;
+
+        pthread_mutex_t   mtx;   /* guards id_to_pf  */
+        pthread_cond_t    cond;
+
+        /* One management thread and one link monitor for all PoAs. */
+        struct llist      mgmt_frames;
+        struct list_head  detach;
+        pthread_mutex_t   mgmt_mtx;
+        pthread_cond_t    mgmt_cond; /* work for the handler   */
+        pthread_cond_t    done_cond; /* a detach has completed */
+        pthread_t         mgmt_handler;
+        bool              mgmt_stop;
+
+        pthread_t         monitor;
+        int               mon_fd;
+
+        enum poa_state    state;
+
+        /* Hashes of the names this process answers queries for. */
+        uint8_t           name_hash[POA_QUERY_HLEN];
+        uint8_t           layer_hash[POA_QUERY_HLEN];
+
+        pthread_rwlock_t  lock;  /* the PoAs and their flows */
+} poas;
+
+static int mgmt_send(struct poa *            poa,
+                     const struct poa_addr * dst,
+                     uint8_t                 code,
+                     uint32_t                s_eid,
+                     uint32_t                d_eid,
+                     qosspec_t               qs,
+                     int                     response,
+                     const buffer_t *        data)
+{
+        uint8_t               buf[POA_MGMT_BUFSZ];
+        struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf;
+        size_t                len;
+
+        len = sizeof(*msg);
+        if (data != NULL && data->len > 0) {
+                if (len + data->len > sizeof(buf))
+                        return -EMSGSIZE;
+                memcpy(buf + len, data->data, data->len);
+
+                len += data->len;
+        }
+
+        poa_mgmt_msg_ser(msg, code, s_eid, d_eid, qs, response,
+                         data != NULL ? data->len : 0);
+
+        POA_STAT_BUMP(poa, mgmt_tx);
+
+        return poa->ops->poa_send_mgmt(poa, dst, buf, len);
+}
+
+/* Caller holds poas.lock for writing. */
+static struct poa_flow * pf_create(struct poa *            poa,
+                                   const struct poa_addr * r_addr)
+{
+        struct poa_flow *  pf;
+        pthread_condattr_t cattr;
+        int                eid;
+
+        pf = malloc(sizeof(*pf));
+        if (pf == NULL)
+                goto fail_malloc;
+
+        memset(pf, 0, sizeof(*pf));
+
+        if (pthread_mutex_init(&pf->mtx, NULL) != 0)
+                goto fail_mtx;
+
+        if (pthread_condattr_init(&cattr) != 0)
+                goto fail_cond;
+#ifndef __APPLE__
+        pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+        if (pthread_cond_init(&pf->cond, &cattr) != 0) {
+                pthread_condattr_destroy(&cattr);
+                goto fail_cond;
+        }
+
+        pthread_condattr_destroy(&cattr);
+
+        eid = bmp_allocate(poa->eids);
+        if (!bmp_is_id_valid(poa->eids, eid))
+                goto fail_eid;
+
+        pf->poa     = poa;
+        pf->eid     = (uint32_t) eid;
+        pf->flow_id = -1;
+        pf->state   = POA_FLOW_PENDING;
+        pf->r_addr  = *r_addr;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &pf->t0);
+
+        rcu_wrlock(&poa->guard);
+
+        rcu_publish(pf);
+        rcu_assign(poa->eid_to_pf[eid], pf);
+
+        rcu_wrunlock(&poa->guard);
+
+        list_add_tail(&pf->pend, &poa->flows);
+
+        POA_STAT_BUMP(poa, n_flows);
+
+        return pf;
+
+ fail_eid:
+        pthread_cond_destroy(&pf->cond);
+ fail_cond:
+        pthread_mutex_destroy(&pf->mtx);
+ fail_mtx:
+        free(pf);
+ fail_malloc:
+        return NULL;
+}
+
+/* Caller holds poas.mtx. */
+static void deferred_purge(int flow_id)
+{
+        struct list_head * p;
+        struct list_head * h;
+
+        list_for_each_safe(p, h, &poas.deferred.list) {
+                struct poa_deferred * d;
+
+                d = list_entry(p, struct poa_deferred, next);
+                if (d->flow_id != flow_id)
+                        continue;
+
+                llist_del(&d->next, &poas.deferred);
+                freebuf(d->data);
+                free(d);
+        }
+}
+
+/* The id may have moved to a newer flow; only its owner clears it. */
+static void pf_destroy(struct poa_flow * pf)
+{
+        struct poa * poa = pf->poa;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        rcu_wrlock(&poa->guard);
+
+        rcu_assign(poa->eid_to_pf[pf->eid], NULL);
+        rcu_assign(pf->rx_rb, NULL);
+
+        rcu_wrunlock(&poa->guard);
+
+        bmp_release(poa->eids, pf->eid);
+
+        if (!list_is_empty(&pf->pend)) {
+                list_del(&pf->pend);
+                POA_STAT_SUB(poa, n_flows, 1);
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        pthread_mutex_lock(&poas.mtx);
+
+        if (pf->flow_id >= 0 && poas.id_to_pf[pf->flow_id] == pf) {
+                poas.id_to_pf[pf->flow_id] = NULL;
+
+                deferred_purge(pf->flow_id);
+        }
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        rcu_reclaim(&poa->guard); /* a receive may still hold pf */
+
+        freebuf(pf->resp_data);
+        freebuf(pf->answer_data);
+
+        pthread_cond_destroy(&pf->cond);
+        pthread_mutex_destroy(&pf->mtx);
+
+        free(pf);
+}
+
+/*
+ * Keeps the answer for a repeat, then sends it. An accept must wait
+ * until the flow can receive; a refusal needs no receiver.
+ */
+static int pf_answer(struct poa_flow * pf,
+                     int               response,
+                     const buffer_t *  data)
+{
+        int err;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        freebuf(pf->answer_data);
+
+        if (data != NULL && data->len > 0) {
+                pf->answer_data.data = malloc(data->len);
+                if (pf->answer_data.data != NULL) {
+                        memcpy(pf->answer_data.data, data->data, data->len);
+
+                        pf->answer_data.len = data->len;
+                }
+        }
+
+        pf->answer   = response;
+        pf->answered = true;
+        if (response == 0 && pf->state != POA_FLOW_ALLOCATED) {
+                pthread_rwlock_unlock(&poas.lock);
+                return 0;
+        }
+
+        pf->answer_sent = true;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        err = mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid,
+                        pf->r_eid, qos_raw, response, data);
+
+        if (err == -ETIMEDOUT || err == -EAGAIN)
+                err = 0; /* stored; a repeat request resends it */
+
+        return err;
+}
+
+/* Takes an answer left for a flow that had no id yet. */
+static struct poa_deferred * deferred_take(int flow_id)
+{
+        struct list_head * p;
+        struct list_head * h;
+
+        list_for_each_safe(p, h, &poas.deferred.list) {
+                struct poa_deferred * d;
+
+                d = list_entry(p, struct poa_deferred, next);
+                if (d->flow_id != flow_id)
+                        continue;
+
+                llist_del(&d->next, &poas.deferred);
+
+                return d;
+        }
+
+        return NULL;
+}
+
+/* Publishes the flow_id so an answer can find this flow. */
+static void pf_set_flow_id(struct poa_flow * pf,
+                           int               flow_id)
+{
+        struct poa_deferred * d;
+
+        pthread_mutex_lock(&poas.mtx);
+
+        pf->flow_id          = flow_id;
+        poas.id_to_pf[flow_id] = pf;
+
+        d = deferred_take(flow_id);
+
+        pthread_cond_broadcast(&poas.cond);
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        if (d != NULL) {
+                pf_answer(pf, d->response, &d->data);
+                freebuf(d->data);
+                free(d);
+        }
+}
+
+/*
+ * Between the request arriving and the accept returning, the flow has
+ * an id but no fd yet; flow_init claims the PoA here.
+ */
+static void pf_set_pending(struct poa_flow * pf)
+{
+        pthread_mutex_lock(&poas.mtx);
+
+        pf->pending = true;
+
+        pthread_mutex_unlock(&poas.mtx);
+}
+
+static void pf_clr_pending(struct poa_flow * pf)
+{
+        pthread_mutex_lock(&poas.mtx);
+
+        pf->pending = false;
+
+        pthread_mutex_unlock(&poas.mtx);
+}
+
+/* A process that attaches no PoA has nothing pending. */
+struct poa_flow * poa_flow_take_pending(int flow_id)
+{
+        struct poa_flow * pf;
+
+        if (poas.state == POA_NULL)
+                return NULL;
+
+        if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+                return NULL;
+
+        pthread_mutex_lock(&poas.mtx);
+
+        pf = poas.id_to_pf[flow_id];
+        if (pf != NULL && pf->pending)
+                pf->pending = false;
+        else
+                pf = NULL;
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        return pf;
+}
+
+static struct poa_flow * pf_get(int flow_id)
+{
+        struct poa_flow * pf;
+
+        if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+                return NULL;
+
+        pthread_mutex_lock(&poas.mtx);
+
+        pf = poas.id_to_pf[flow_id];
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        return pf;
+}
+
+#ifdef PROC_FLOW_STATS
+
+/* Caller holds poas.lock. */
+static struct poa * poa_by_rib_name(const char * name)
+{
+        struct list_head * p;
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (strcmp(poa->name, name) == 0)
+                        return poa;
+        }
+
+        return NULL;
+}
+
+static int poa_rib_read(const char * path,
+                        char *       buf,
+                        size_t       len)
+{
+        struct poa * poa;
+        const char * entry;
+        size_t       qlen;
+        size_t       avg;
+        size_t       cost;
+        int          size;
+        int          ret;
+
+        entry = strstr(path, RIB_SEPARATOR) + 1;
+
+        if (len < POA_RIB_STRLEN)
+                return -1;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        poa = poa_by_rib_name(entry);
+        if (poa == NULL)
+                goto fail;
+
+        qlen = poa->ops->poa_qlen(poa);
+        avg  = poa->avg_len;
+        cost = poa->avg_len + poa->avg_ovh;
+
+        size = snprintf(buf, len,
+                        "Active flows:              %zu\n"
+                        "Packets received:          %zu\n"
+                        "Bytes received:            %zu\n"
+                        "Packets sent:              %zu\n"
+                        "Bytes sent:                %zu\n"
+                        "Management frames rcvd:    %zu\n"
+                        "Management frames sent:    %zu\n"
+                        "Bad EID packets:           %zu\n"
+                        "Delivery (N+1) failures:   %zu\n"
+                        "Buffer alloc failures:     %zu\n"
+                        "Packet read failures:      %zu\n"
+                        "Packet send failures:      %zu\n"
+                        "Name queries sent:         %zu\n"
+                        "Name queries received:     %zu\n"
+                        "Name replies sent:         %zu\n"
+                        "Name replies received:     %zu\n"
+                        "Queued (transport):        %zu\n"
+                        "Queued (packets):          %zu\n"
+                        "Mean packet size (bytes):  %zu\n"
+                        "Mean packet cost:          %zu\n",
+                        POA_STAT_LOAD(poa, n_flows),
+                        POA_STAT_LOAD(poa, rx_pkts),
+                        POA_STAT_LOAD(poa, rx_bytes),
+                        POA_STAT_LOAD(poa, tx_pkts),
+                        POA_STAT_LOAD(poa, tx_bytes),
+                        POA_STAT_LOAD(poa, mgmt_rx),
+                        POA_STAT_LOAD(poa, mgmt_tx),
+                        POA_STAT_LOAD(poa, bad_eid),
+                        POA_STAT_LOAD(poa, dlv_fail),
+                        POA_STAT_LOAD(poa, buf_fail),
+                        POA_STAT_LOAD(poa, rcv_fail),
+                        POA_STAT_LOAD(poa, snd_fail),
+                        POA_STAT_LOAD(poa, qry_tx),
+                        POA_STAT_LOAD(poa, qry_rx),
+                        POA_STAT_LOAD(poa, rep_tx),
+                        POA_STAT_LOAD(poa, rep_rx),
+                        qlen, cost > 0 ? qlen / cost : 0, avg, cost);
+        if (size < 0 || (size_t) size >= len)
+                goto fail;
+
+        if (poa->ops->poa_rib != NULL) {
+                ret = poa->ops->poa_rib(poa, buf + size, len - size);
+                if (ret < 0)
+                        goto fail;
+
+                size += ret;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return size;
+
+ fail:
+        pthread_rwlock_unlock(&poas.lock);
+
+        return -1;
+}
+
+static int poa_rib_readdir(char *** buf)
+{
+        struct list_head * p;
+        size_t             n = 0;
+        int                idx = 0;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list)
+                ++n;
+
+        if (n == 0) {
+                *buf = NULL;
+                goto no_poas;
+        }
+
+        *buf = malloc(sizeof(**buf) * n);
+        if (*buf == NULL)
+                goto fail_entries;
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                (*buf)[idx] = malloc(strlen(poa->name) + 1);
+                if ((*buf)[idx] == NULL)
+                        goto fail_entry;
+
+                strcpy((*buf)[idx++], poa->name);
+        }
+ no_poas:
+        pthread_rwlock_unlock(&poas.lock);
+
+        return idx;
+
+ fail_entry:
+        while (idx-- > 0)
+                free((*buf)[idx]);
+
+        free(*buf);
+ fail_entries:
+        pthread_rwlock_unlock(&poas.lock);
+
+        return -ENOMEM;
+}
+
+static int poa_rib_getattr(const char *      path,
+                           struct rib_attr * attr)
+{
+        (void) path;
+
+        attr->size  = POA_RIB_STRLEN;
+        attr->mtime = 0;
+
+        return 0;
+}
+
+static struct rib_ops poa_r_ops = {
+        .read    = poa_rib_read,
+        .readdir = poa_rib_readdir,
+        .getattr = poa_rib_getattr
+};
+
+#endif /* PROC_FLOW_STATS */
+
+int poa_init(const char * name)
+{
+        pthread_condattr_t cattr;
+
+        assert(name != NULL);
+
+        memset(&poas, 0, sizeof(poas));
+
+        str_hash(HASH_SHA3_256, poas.name_hash, name);
+
+        poas.mon_fd = -1;
+        if (pthread_mutex_init(&poas.mtx, NULL) != 0)
+                goto fail_mtx;
+
+        if (pthread_cond_init(&poas.cond, NULL) != 0)
+                goto fail_cond;
+
+        if (pthread_mutex_init(&poas.mgmt_mtx, NULL) != 0)
+                goto fail_mgmt_mtx;
+
+        if (pthread_condattr_init(&cattr) != 0)
+                goto fail_cattr;
+#ifndef __APPLE__
+        pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK);
+#endif
+        if (pthread_cond_init(&poas.mgmt_cond, &cattr) != 0) {
+                pthread_condattr_destroy(&cattr);
+                goto fail_cattr;
+        }
+
+        pthread_condattr_destroy(&cattr);
+
+        if (pthread_cond_init(&poas.done_cond, NULL) != 0)
+                goto fail_done_cond;
+
+        if (pthread_rwlock_init(&poas.lock, NULL) != 0)
+                goto fail_lock;
+
+        list_head_init(&poas.list);
+        llist_init(&poas.deferred);
+        llist_init(&poas.mgmt_frames);
+        list_head_init(&poas.detach);
+
+        poas.state = POA_INIT;
+
+#ifdef PROC_FLOW_STATS
+        if (rib_reg(POA_RIB, &poa_r_ops) < 0)
+                goto fail_rib;
+#endif
+        return 0;
+
+#ifdef PROC_FLOW_STATS
+ fail_rib:
+        pthread_rwlock_destroy(&poas.lock);
+#endif
+
+ fail_lock:
+        pthread_cond_destroy(&poas.done_cond);
+ fail_done_cond:
+        pthread_cond_destroy(&poas.mgmt_cond);
+ fail_cattr:
+        pthread_mutex_destroy(&poas.mgmt_mtx);
+ fail_mgmt_mtx:
+        pthread_cond_destroy(&poas.cond);
+ fail_cond:
+        pthread_mutex_destroy(&poas.mtx);
+ fail_mtx:
+        return -1;
+}
+
+int poa_set_layer(const char * layer)
+{
+        if (layer == NULL)
+                return -EINVAL;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        str_hash(HASH_SHA3_256, poas.layer_hash, layer);
+
+        poas.state = POA_OPERATIONAL;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return 0;
+}
+
+bool poa_has_name(const uint8_t * hash)
+{
+        bool match = false;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        if (poas.state >= POA_INIT)
+                match = memcmp(hash, poas.name_hash, POA_QUERY_HLEN) == 0;
+
+        if (!match && poas.state >= POA_OPERATIONAL)
+                match = memcmp(hash, poas.layer_hash, POA_QUERY_HLEN) == 0;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return match;
+}
+
+int poa_spb_reserve(struct ssm_pk_buff ** spb,
+                    size_t                len)
+{
+        return ssm_pool_alloc_b(proc.pool, len, NULL, spb, NULL) < 0 ? -1 : 0;
+}
+
+void poa_spb_release(struct ssm_pk_buff * spb)
+{
+        ssm_pool_remove(proc.pool, ssm_pk_buff_get_off(spb));
+}
+
+void poa_rx_pkt(struct poa *         poa,
+                uint32_t             eid,
+                struct ssm_pk_buff * spb)
+{
+        struct poa_flow *  pf;
+        struct ssm_rbuff * rx_rb;
+        size_t             len;
+
+        len = ssm_pk_buff_len(spb); /* the ring write takes it over */
+
+        if (eid >= poa->n_eids) {
+                POA_STAT_BUMP(poa, bad_eid);
+                poa_spb_release(spb);
+                return;
+        }
+
+        rcu_rdlock(&poa->guard);
+
+        pf = rcu_deref(poa->eid_to_pf[eid]);
+
+        rcu_consume(pf);
+
+        if (pf == NULL)
+                goto fail_eid;
+
+        rx_rb = rcu_deref(pf->rx_rb);
+
+        rcu_consume(rx_rb);
+
+        if (rx_rb == NULL)
+                goto fail_eid;
+
+        if (ssm_rbuff_write(rx_rb, ssm_pk_buff_get_off(spb)) < 0) {
+                POA_STAT_BUMP(poa, dlv_fail);
+                rcu_rdunlock(&poa->guard);
+                poa_spb_release(spb);
+                return;
+        }
+
+        POA_STAT_BUMP(poa, rx_pkts);
+        POA_STAT_ADD(poa, rx_bytes, len);
+
+        ssm_flow_set_notify(proc.fqset, pf->flow_id, FLOW_PKT);
+
+        rcu_rdunlock(&poa->guard);
+
+        return;
+
+ fail_eid:
+        POA_STAT_BUMP(poa, bad_eid);
+
+        rcu_rdunlock(&poa->guard);
+
+        poa_spb_release(spb);
+}
+
+static int poa_flow_req_arr(struct poa_flow * pf,
+                            qosspec_t         qs,
+                            uint32_t          mtu,
+                            const buffer_t *  data)
+{
+        struct flow_info flow;
+        uint8_t          buf[SOCK_BUF_SIZE];
+        buffer_t         msg = {SOCK_BUF_SIZE, buf};
+        buffer_t         out = BUF_INIT;
+        int              err;
+
+        memset(&flow, 0, sizeof(flow));
+
+        flow.n_pid   = getpid();
+        flow.n_1_pid = getpid();
+        flow.qs      = qs;
+        flow.mpl     = pf->poa->mpl;
+        flow.mtu     = mtu;
+        if (ipcp_poa_flow_req_arr__irm_req_ser(&msg, &flow, data) < 0)
+                return -ENOMEM;
+
+        pf_set_pending(pf);
+
+        err = send_recv_msg(&msg);
+        if (err < 0)
+                goto fail;
+
+        err = poa_flow__irm_result_des(&msg, &flow, &out);
+        if (err < 0)
+                goto fail;
+
+        freebuf(out);
+
+        if (flow.id < 0 || flow.id >= SYS_MAX_FLOWS) {
+                err = -EBADF;
+                goto fail;
+        }
+
+        pf_set_flow_id(pf, flow.id);
+
+        return 0;
+ fail:
+        pf_clr_pending(pf);
+        return err;
+}
+
+static void handle_flow_req(struct poa *                poa,
+                            const struct poa_addr *     src,
+                            const struct poa_mgmt_msg * msg,
+                            const uint8_t *             data,
+                            size_t                      data_len)
+{
+        struct list_head * p;
+        struct poa_flow *  pf     = NULL;
+        qosspec_t          qs;
+        buffer_t           buf;
+        buffer_t           answer_data = BUF_INIT;
+        uint32_t           r_eid;
+        uint32_t           mtu;
+        uint32_t           eid    = 0;
+        int                answer = 0;
+        bool               found  = false;
+        bool               served = false;
+
+        r_eid = ntoh32(msg->s_eid);
+
+        poa_mgmt_msg_qos(msg, &qs);
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        list_for_each(p, &poa->flows) {
+                pf = list_entry(p, struct poa_flow, pend);
+                if (pf->r_eid != r_eid || poa_addr_cmp(&pf->r_addr, src) != 0)
+                        continue;
+
+                answer = pf->answer;
+                eid    = pf->eid;
+                served = true;
+
+                found = pf->answer_sent;
+                if (found && pf->answer_data.len > 0) {
+                        answer_data.data = malloc(pf->answer_data.len);
+                        if (answer_data.data != NULL) {
+                                memcpy(answer_data.data, pf->answer_data.data,
+                                       pf->answer_data.len);
+                                answer_data.len = pf->answer_data.len;
+                        }
+                }
+                break;
+        }
+
+        if (!served) {
+                pf = pf_create(poa, src);
+                if (pf != NULL)
+                        pf->r_eid = r_eid;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        if (found) {
+                mgmt_send(poa, src, POA_FLOW_REPLY, eid, r_eid, qos_raw, answer,
+                          answer_data.len > 0 ? &answer_data : NULL);
+                freebuf(answer_data);
+        }
+
+        if (served || pf == NULL)
+                return;
+
+        buf.len  = data_len;
+        buf.data = (uint8_t *) data;
+        mtu      = poa->ops->poa_mtu(poa, src);
+        if (poa_flow_req_arr(pf, qs, mtu, &buf) < 0) {
+                mgmt_send(poa, src, POA_FLOW_REPLY, pf->eid, r_eid, qos_raw,
+                          -1, NULL);
+                pf_destroy(pf);
+        }
+}
+
+static void handle_flow_reply(struct poa *                poa,
+                              const struct poa_addr *     src,
+                              const struct poa_mgmt_msg * msg,
+                              const uint8_t *             data,
+                              size_t                      data_len)
+{
+        struct poa_flow * pf;
+        uint32_t          eid;
+
+        eid = ntoh32(msg->d_eid);
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL;
+        if (pf == NULL || pf->state != POA_FLOW_PENDING) {
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        if (poa_addr_cmp(&pf->r_addr, src) != 0) {
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        pthread_mutex_lock(&pf->mtx);
+
+        if (pf->replied) {
+                pthread_mutex_unlock(&pf->mtx);
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        if (data_len > 0) {
+                pf->resp_data.data = malloc(data_len);
+                if (pf->resp_data.data != NULL) {
+                        memcpy(pf->resp_data.data, data, data_len);
+
+                        pf->resp_data.len = data_len;
+                }
+        }
+
+        pf->r_eid    = ntoh32(msg->s_eid);
+        pf->response = ntoh32(msg->response);
+        pf->replied  = true;
+
+        pthread_cond_broadcast(&pf->cond);
+
+        pthread_mutex_unlock(&pf->mtx);
+
+        pthread_rwlock_unlock(&poas.lock);
+}
+
+static void handle_flow_update(struct poa *                poa,
+                               const struct poa_addr *     src,
+                               const struct poa_mgmt_msg * msg,
+                               const uint8_t *             data,
+                               size_t                      data_len)
+{
+        struct poa_flow * pf;
+        buffer_t          buf;
+        uint32_t          eid;
+
+        eid = ntoh32(msg->d_eid);
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL;
+        if (pf == NULL || pf->state != POA_FLOW_ALLOCATED) {
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        if (poa_addr_cmp(&pf->r_addr, src) != 0) {
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        eid = (uint32_t) pf->flow_id;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        buf.len  = data_len;
+        buf.data = (uint8_t *) data;
+
+        ipcp_flow_update_arr((int) eid, &buf);
+}
+
+static void mgmt_frame_handle(struct poa_mgmt_frame * frame)
+{
+        const struct poa_mgmt_msg * msg;
+        const uint8_t *             data;
+        size_t                      data_len;
+
+        msg = (const struct poa_mgmt_msg *) frame->buf;
+        if (frame->len < sizeof(*msg))
+                return;
+
+        data_len = ntoh16(msg->data_len);
+        if (data_len > frame->len - sizeof(*msg))
+                return;
+
+        data = frame->buf + sizeof(*msg);
+
+        switch (msg->code) {
+        case POA_FLOW_REQ:
+                handle_flow_req(frame->poa, &frame->src, msg, data, data_len);
+                break;
+        case POA_FLOW_REPLY:
+                handle_flow_reply(frame->poa, &frame->src, msg, data, data_len);
+                break;
+        case POA_FLOW_UPDATE:
+                handle_flow_update(frame->poa, &frame->src, msg, data,
+                                   data_len);
+                break;
+        default:
+                break;
+        }
+}
+
+static bool pf_steal(struct poa_flow * pf)
+{
+        bool stolen = false;
+
+        pthread_mutex_lock(&poas.mtx);
+
+        if (pf->pending) {
+                pf->pending = false;
+                poas.id_to_pf[pf->flow_id] = NULL;
+
+                deferred_purge(pf->flow_id);
+
+                stolen = true;
+        }
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        return stolen;
+}
+
+static void sweep_pending(void)
+{
+        struct poa_flow *  dead[POA_SWEEP_MAX];
+        struct list_head * p;
+        struct list_head * q;
+        struct timespec    now;
+        size_t             n = 0;
+        size_t             i;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                list_for_each(q, &poa->flows) {
+                        struct poa_flow * pf;
+
+                        if (n == POA_SWEEP_MAX)
+                                break;
+
+                        pf = list_entry(q, struct poa_flow, pend);
+                        if (pf->state != POA_FLOW_PENDING || pf->claimed)
+                                continue;
+
+                        if (now.tv_sec - pf->t0.tv_sec < POA_PEND_TIMEO)
+                                continue;
+
+                        if (pf->flow_id >= 0 && !pf_steal(pf))
+                                continue;
+
+                        dead[n++] = pf;
+                }
+
+                if (n == POA_SWEEP_MAX)
+                        break;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        for (i = 0; i < n; ++i)
+                pf_destroy(dead[i]);
+
+        pthread_mutex_lock(&poas.mtx);
+
+        list_for_each_safe(p, q, &poas.deferred.list) {
+                struct poa_deferred * d;
+
+                d = list_entry(p, struct poa_deferred, next);
+                if (now.tv_sec - d->t0.tv_sec < POA_PEND_TIMEO)
+                        continue;
+
+                llist_del(&d->next, &poas.deferred);
+                freebuf(d->data);
+                free(d);
+        }
+
+        pthread_mutex_unlock(&poas.mtx);
+}
+
+void poa_rx_mgmt(struct poa *            poa,
+                 const struct poa_addr * src,
+                 const uint8_t *         buf,
+                 size_t                  len)
+{
+        struct poa_mgmt_frame * frame;
+
+        if (len < sizeof(struct poa_mgmt_msg) || len > POA_MGMT_BUFSZ)
+                return;
+
+        POA_STAT_BUMP(poa, mgmt_rx);
+
+        frame = malloc(offsetof(struct poa_mgmt_frame, buf) + len);
+        if (frame == NULL)
+                return;
+
+        frame->poa = poa;
+        frame->src = *src;
+        frame->len = len;
+
+        memcpy(frame->buf, buf, len);
+
+        pthread_mutex_lock(&poas.mgmt_mtx);
+
+        if (poa->n_mgmt >= POA_MGMT_QMAX) {
+                pthread_mutex_unlock(&poas.mgmt_mtx);
+                free(frame);
+                return;
+        }
+
+        ++poa->n_mgmt;
+
+        llist_add_tail(&frame->next, &poas.mgmt_frames);
+
+        pthread_cond_signal(&poas.mgmt_cond);
+
+        pthread_mutex_unlock(&poas.mgmt_mtx);
+}
+
+static void mgmt_frames_purge(const struct poa * poa)
+{
+        struct list_head * p;
+        struct list_head * h;
+
+        pthread_mutex_lock(&poas.mgmt_mtx);
+
+        list_for_each_safe(p, h, &poas.mgmt_frames.list) {
+                struct poa_mgmt_frame * frame;
+
+                frame = list_entry(p, struct poa_mgmt_frame, next);
+                if (frame->poa != poa)
+                        continue;
+
+                --frame->poa->n_mgmt;
+
+                llist_del(&frame->next, &poas.mgmt_frames);
+
+                free(frame);
+        }
+
+        pthread_mutex_unlock(&poas.mgmt_mtx);
+}
+
+static bool poa_has_id(const struct poa *      poa,
+                       const struct poa_spec * spec)
+{
+        if (poa->type != spec->type)
+                return false;
+
+        return poa->ops->poa_has_id(poa, spec);
+}
+
+/*
+ * The PoA carrying dst, as the transport judges it. -EPERM if none
+ * matches, -EINVAL if several do. Caller holds poas.lock.
+ */
+static int poa_lookup(const struct poa_addr * dst,
+                      struct poa **           out)
+{
+        struct list_head * p;
+        struct poa *       found = NULL;
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->type != dst->type)
+                        continue;
+
+                if (!poa->ops->poa_match(poa, dst))
+                        continue;
+
+                if (found != NULL) /* nothing given, two candidates */
+                        return -EINVAL;
+
+                found = poa;
+        }
+
+        if (found == NULL)
+                return -EPERM;
+
+        *out = found;
+
+        return 0;
+}
+
+/* Sends are deadlined, bounding the lock hold on a full queue. */
+int poa_bcast_mgmt(const struct poa_addr * dst,
+                   const uint8_t *         buf,
+                   size_t                  len)
+{
+        struct list_head * p;
+        int                n = 0;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->type != dst->type)
+                        continue;
+
+                if (!poa->ops->poa_match(poa, dst))
+                        continue;
+
+                if (poa->ops->poa_send_mgmt(poa, dst, buf, len) < 0)
+                        continue;
+
+                /* All management broadcasts are name queries. */
+                POA_STAT_BUMP(poa, qry_tx);
+                ++n;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return n;
+}
+
+static bool deadline_is_malformed(const struct timespec * timeo)
+{
+        if (timeo == NULL)
+                return false;
+
+        if (timeo->tv_sec < 0 || timeo->tv_nsec < 0)
+                return true;
+
+        return timeo->tv_nsec >= BILLION;
+}
+
+/*
+ * Complete addr for dst on any backend that can query. The ops are
+ * collected under poas.lock but called outside it: a query blocks up
+ * to its deadline and takes the lock again to broadcast. The tables
+ * are static, so nothing dangles; a struct poa cannot be carried
+ * across the unlock. The deadline applies per backend.
+ */
+int poa_query(const char *            dst,
+              const struct timespec * timeo,
+              struct poa_addr *       addr)
+{
+        const struct poa_ops * cand[POA_MAX_POAS];
+        enum poa_type          type[POA_MAX_POAS];
+        struct list_head *     p;
+        size_t                 n = 0;
+        size_t                 i;
+        int                    err = -ENOTSUP;
+
+        if (dst == NULL || addr == NULL)
+                return -EINVAL;
+
+        if (deadline_is_malformed(timeo))
+                return -EINVAL;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->ops->poa_query == NULL)
+                        continue;
+
+                for (i = 0; i < n && cand[i] != poa->ops; i++)
+                        ;
+                if (i < n)
+                        continue;
+
+                /* One type per backend: eth. Revisit if that changes. */
+                cand[n]   = poa->ops;
+                type[n++] = poa->type;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        for (i = 0; i < n; i++) {
+                memset(addr, 0, sizeof(*addr));
+
+                addr->type = type[i];
+
+                err = cand[i]->poa_query(dst, timeo, addr);
+                if (err == 0)
+                        return 0;
+        }
+
+        return err;
+}
+
+static int poa_check(const struct poa_addr * dst)
+{
+        struct poa * poa;
+        int          err;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        err = poa_lookup(dst, &poa);
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return err;
+}
+
+static int poa_alloc(const struct poa_addr * dst,
+                     qosspec_t               qs,
+                     const buffer_t *        req,
+                     buffer_t *              resp,
+                     struct poa_flow **      pf_out,
+                     uint32_t *              mtu,
+                     const struct timespec * timeo)
+{
+        struct timespec   dflt  = TIMESPEC_INIT_MS(POA_ALLOC_TIMEO);
+        struct timespec   rintv = TIMESPEC_INIT_MS(POA_RETRY_TIMEO);
+        struct poa_flow * pf;
+        struct poa *      poa;
+        struct timespec   abstime;
+        struct timespec   now;
+        struct timespec   retry;
+        int               err;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        err = poa_lookup(dst, &poa);
+        if (err < 0) {
+                pthread_rwlock_unlock(&poas.lock);
+                return err;
+        }
+
+        pf = pf_create(poa, dst);
+        if (pf == NULL) {
+                pthread_rwlock_unlock(&poas.lock);
+                return -ENOMEM;
+        }
+
+        pf->claimed = true;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+        ts_add(&abstime, timeo != NULL ? timeo : &dflt, &abstime);
+
+        pthread_mutex_lock(&pf->mtx);
+
+        while (!pf->replied) {
+                pthread_mutex_unlock(&pf->mtx);
+
+                err = mgmt_send(poa, dst, POA_FLOW_REQ, pf->eid, 0, qs, 0, req);
+
+                pthread_mutex_lock(&pf->mtx);
+
+                if (err < 0 && err != -ETIMEDOUT && err != -EAGAIN) {
+                        err = -EIO;
+                        goto fail;
+                }
+
+                if (pf->replied)
+                        break;
+
+                clock_gettime(PTHREAD_COND_CLOCK, &now);
+                ts_add(&now, &rintv, &retry);
+
+                if (ts_diff_ns(&retry, &abstime) > 0)
+                        retry = abstime;
+
+                pthread_cond_timedwait(&pf->cond, &pf->mtx, &retry);
+
+                if (pf->replied)
+                        break;
+
+                clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+                if (ts_diff_ns(&now, &abstime) >= 0) {
+                        err = -ETIMEDOUT;
+                        goto fail;
+                }
+        }
+
+        if (pf->response != 0) {
+                err = -ECONNREFUSED;
+                goto fail;
+        }
+
+        *resp              = pf->resp_data;
+        pf->resp_data.len  = 0;
+        pf->resp_data.data = NULL;
+
+        pthread_mutex_unlock(&pf->mtx);
+
+        *mtu    = poa->ops->poa_mtu(poa, dst);
+        *pf_out = pf;
+
+        return 0;
+
+ fail:
+        pthread_mutex_unlock(&pf->mtx);
+        pf_destroy(pf);
+        return err;
+}
+
+static void poa_alloc_fail(struct poa_flow * pf)
+{
+        pf_destroy(pf);
+}
+
+void poa_flow_attach(struct poa_flow *  pf,
+                     int                flow_id,
+                     struct ssm_rbuff * rx_rb)
+{
+        struct poa * poa = pf->poa;
+
+        if (pf->flow_id != flow_id)
+                pf_set_flow_id(pf, flow_id);
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        pf->state = POA_FLOW_ALLOCATED;
+
+        rcu_wrlock(&poa->guard);
+
+        rcu_publish(rx_rb);
+        rcu_assign(pf->rx_rb, rx_rb);
+
+        rcu_wrunlock(&poa->guard);
+
+        pthread_rwlock_unlock(&poas.lock);
+}
+
+void poa_flow_detach(struct poa_flow * pf)
+{
+        struct poa * poa = pf->poa;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        pf->state = POA_FLOW_DEAD;
+
+        rcu_wrlock(&poa->guard);
+
+        rcu_assign(pf->rx_rb, NULL);
+
+        rcu_wrunlock(&poa->guard);
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        pf_destroy(pf);
+}
+
+static size_t flows_updown(struct poa * poa,
+                           bool         up)
+{
+        struct list_head * p;
+        size_t             n = 0;
+
+        list_for_each(p, &poa->flows) {
+                struct poa_flow * pf;
+                struct flow *     flow;
+
+                pf = list_entry(p, struct poa_flow, pend);
+                if (pf->state != POA_FLOW_ALLOCATED || pf->flow_id < 0)
+                        continue;
+
+                flow = &proc.flows[proc.id_to_fd[pf->flow_id].fd];
+                if (flow->info.id != pf->flow_id)
+                        continue;
+
+                if (((flow->oflags & FLOWFDOWN) != 0) != !up)
+                        ++n;
+
+                if (up) {
+                        flow->oflags &= ~FLOWFDOWN;
+
+                        ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN);
+                } else {
+                        flow->oflags |= FLOWFDOWN;
+
+                        ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN);
+                }
+
+                ssm_flow_set_notify(proc.fqset, pf->flow_id,
+                                    up ? FLOW_UP : FLOW_DOWN);
+        }
+
+        return n;
+}
+
+size_t poa_link_updown(int  id,
+                       bool up)
+{
+        struct list_head * p;
+        size_t             n = 0;
+
+        pthread_rwlock_wrlock(&proc.lock);
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->ops->poa_link_match == NULL)
+                        continue;
+
+                if (!poa->ops->poa_link_match(poa, id))
+                        continue;
+
+                n += flows_updown(poa, up);
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+        pthread_rwlock_unlock(&proc.lock);
+
+        return n;
+}
+
+static size_t poa_ewma(size_t avg,
+                       size_t sz)
+{
+        if (avg == 0)
+                return sz;
+
+        avg = avg + (sz >> POA_AVG_SHIFT) - (avg >> POA_AVG_SHIFT);
+
+        return avg == 0 ? 1 : avg;
+}
+
+static void poa_avg_len_update(struct poa * poa,
+                               size_t       sz)
+{
+        STORE_RELAXED(&poa->avg_len, poa_ewma(LOAD_RELAXED(&poa->avg_len), sz));
+}
+
+static size_t poa_qlen(struct poa * poa)
+{
+        struct timespec now;
+        uint64_t        ns;
+        size_t          qlen;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+        ns = TS_TO_UINT64(now);
+        if (ns - LOAD_RELAXED(&poa->q_time) < POA_QLEN_GATE)
+                return LOAD_RELAXED(&poa->q_cache);
+
+        qlen = poa->ops->poa_qlen(poa);
+
+        STORE_RELAXED(&poa->q_cache, qlen);
+        STORE_RELAXED(&poa->q_time, ns);
+
+        return qlen;
+}
+
+static void poa_cost_sample(struct poa * poa,
+                            size_t       before,
+                            size_t       len)
+{
+        size_t after;
+
+        after = poa->ops->poa_qlen(poa);
+        if (after <= before)
+                return;   /* drained; nothing to learn */
+
+        after -= before;
+        if (after < len || after > (len << 2) + 1024)
+                return;
+
+        STORE_RELAXED(&poa->avg_ovh,
+                      poa_ewma(LOAD_RELAXED(&poa->avg_ovh), after - len));
+}
+
+int poa_flow_tx(struct poa_flow *       pf,
+                struct ssm_pk_buff *    spb,
+                bool                    block,
+                const struct timespec * abstime)
+{
+        struct poa * poa    = pf->poa;
+        size_t       len    = ssm_pk_buff_len(spb);
+        size_t       before = 0;
+        bool         sample;
+        int          ret;
+
+        sample = (LOAD_RELAXED(&poa->n_tx) & POA_COST_MASK) == 0;
+        if (sample)
+                before = poa->ops->poa_qlen(poa);
+
+        ret = poa->ops->poa_send(poa, &pf->r_addr, pf->r_eid, spb, block,
+                                 abstime);
+        if (ret < 0) { /* the caller releases the buffer */
+                POA_STAT_BUMP(poa, snd_fail);
+                return ret;
+        }
+
+        POA_STAT_BUMP(poa, tx_pkts);
+        POA_STAT_ADD(poa, tx_bytes, len);
+
+        FETCH_ADD_RELAXED(&poa->n_tx, 1);
+
+        poa_avg_len_update(poa, len);
+
+        if (sample)
+                poa_cost_sample(poa, before, len);
+
+        poa_spb_release(spb);
+
+        return 0;
+}
+
+size_t poa_flow_qlen(const struct poa_flow * pf)
+{
+        struct poa * poa = pf->poa;
+        uint64_t     bytes;
+        size_t       cost;
+        size_t       pkts;
+        size_t       byts;
+
+        if (poa->ops->poa_qpkts != NULL &&
+            poa->ops->poa_qpkts(poa, &pkts, &byts) == 0)
+                return byts;
+
+        cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh);
+        if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0)
+                return poa_qlen(poa); /* overstated beats false empty */
+
+        bytes = (uint64_t) poa_qlen(poa) * LOAD_RELAXED(&poa->avg_len);
+
+        return (size_t) (bytes / cost);
+}
+
+size_t poa_flow_qpkts(const struct poa_flow * pf)
+{
+        struct poa * poa = pf->poa;
+        size_t       cost;
+        size_t       pkts;
+        size_t       byts;
+
+        if (poa->ops->poa_qpkts != NULL &&
+            poa->ops->poa_qpkts(poa, &pkts, &byts) == 0)
+                return pkts;
+
+        cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh);
+        if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0)
+                return 0;
+
+        return poa_qlen(poa) / cost;
+}
+
+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)
+{
+        return pf->poa->qid;
+}
+
+void poa_flow_ready(struct poa_flow * pf)
+{
+        buffer_t data;
+        int      answer;
+
+        if (pf == NULL)
+                return;
+
+        clrbuf(data);
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        if (!pf->answered || pf->answer_sent) {
+                pthread_rwlock_unlock(&poas.lock);
+                return;
+        }
+
+        answer = pf->answer;
+
+        if (pf->answer_data.len > 0) {
+                data.data = malloc(pf->answer_data.len);
+                if (data.data != NULL) {
+                        memcpy(data.data, pf->answer_data.data,
+                               pf->answer_data.len);
+                        data.len = pf->answer_data.len;
+                }
+        }
+
+        pf->answer_sent = true;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid,
+                  pf->r_eid, qos_raw, answer, &data);
+
+        freebuf(data);
+}
+
+int poa_flow_alloc_resp(int              flow_id,
+                        int              response,
+                        const buffer_t * data)
+{
+        struct poa_deferred * d;
+        struct poa_flow *     pf;
+
+        if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+                return -EPERM;
+
+        pthread_mutex_lock(&poas.mtx);
+
+        pf = poas.id_to_pf[flow_id];
+        if (pf != NULL) {
+                pthread_mutex_unlock(&poas.mtx);
+                return pf_answer(pf, response, data);
+        }
+
+        if (poas.deferred.len >= POA_DEFER_MAX) {
+                pthread_mutex_unlock(&poas.mtx);
+                return -ENOMEM;
+        }
+
+        d = malloc(sizeof(*d));
+        if (d == NULL) {
+                pthread_mutex_unlock(&poas.mtx);
+                return -ENOMEM;
+        }
+
+        memset(d, 0, sizeof(*d));
+
+        clock_gettime(PTHREAD_COND_CLOCK, &d->t0);
+
+        d->flow_id  = flow_id;
+        d->response = response;
+
+        if (data != NULL && data->len > 0) {
+                d->data.data = malloc(data->len);
+                if (d->data.data == NULL) {
+                        free(d);
+                        pthread_mutex_unlock(&poas.mtx);
+                        return -ENOMEM;
+                }
+                memcpy(d->data.data, data->data, data->len);
+
+                d->data.len = data->len;
+        }
+
+        llist_add_tail(&d->next, &poas.deferred);
+
+        pthread_mutex_unlock(&poas.mtx);
+
+        return 0;
+}
+
+int poa_flow_update(int              flow_id,
+                    const buffer_t * data)
+{
+        struct poa_flow * pf;
+
+        pf = pf_get(flow_id);
+        if (pf == NULL)
+                return -EPERM;
+
+        return mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_UPDATE, pf->eid,
+                         pf->r_eid, qos_raw, 0, data);
+}
+
+/* The PoA state is released when the flow itself is torn down. */
+int poa_flow_dealloc(int flow_id)
+{
+        (void) flow_id;
+
+        return 0;
+}
+
+/* PoA id 0 = management channel. */
+static struct poa * poa_create(enum poa_type          type,
+                               const struct poa_ops * ops,
+                               size_t                 n_eids)
+{
+        struct poa * poa;
+
+        poa = malloc(sizeof(*poa));
+        if (poa == NULL)
+                goto fail_malloc;
+
+        memset(poa, 0, sizeof(*poa));
+
+        poa->eid_to_pf = malloc(sizeof(*poa->eid_to_pf) * n_eids);
+        if (poa->eid_to_pf == NULL)
+                goto fail_map;
+
+        memset(poa->eid_to_pf, 0, sizeof(*poa->eid_to_pf) * n_eids);
+
+        poa->eids = bmp_create(n_eids - 1, 1);
+        if (poa->eids == NULL)
+                goto fail_bmp;
+
+        if (rcu_guard_init(&poa->guard) != 0)
+                goto fail_guard;
+
+        list_head_init(&poa->next);
+        list_head_init(&poa->flows);
+
+        poa->type   = type;
+        poa->ops    = ops;
+        poa->mpl    = ops->mpl;
+        poa->n_eids = n_eids;
+        poa->qid    = -1;
+
+        return poa;
+
+ fail_guard:
+        bmp_destroy(poa->eids);
+ fail_bmp:
+        free(poa->eid_to_pf);
+ fail_map:
+        free(poa);
+ fail_malloc:
+        return NULL;
+}
+
+static void poa_destroy(struct poa * poa)
+{
+        rcu_guard_fini(&poa->guard);
+
+        bmp_destroy(poa->eids);
+
+        free(poa->eid_to_pf);
+        free(poa);
+}
+
+static void poa_teardown(struct poa * poa)
+{
+        if (poas.state >= POA_RUNNING)
+                poa->ops->poa_stop(poa);
+
+        mgmt_frames_purge(poa);
+
+        poa->ops->poa_detach(poa);
+
+        poa_destroy(poa);
+}
+
+static void poa_detach_all(void)
+{
+        pthread_rwlock_wrlock(&poas.lock);
+
+        while (!list_is_empty(&poas.list)) {
+                struct poa * poa;
+
+                poa = list_first_entry(&poas.list, struct poa, next);
+
+                list_del(&poa->next);
+
+                pthread_rwlock_unlock(&poas.lock);
+
+                poa_teardown(poa);
+
+                pthread_rwlock_wrlock(&poas.lock);
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+}
+
+static int poa_do_detach(const struct poa_detach_req * req)
+{
+        struct list_head * p;
+        struct poa *       found = NULL;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (!poa_has_id(poa, &req->spec))
+                        continue;
+
+                found = poa;
+                break;
+        }
+
+        if (found == NULL) {
+                pthread_rwlock_unlock(&poas.lock);
+                return -ENOENT;
+        }
+
+        if (!list_is_empty(&found->flows)) {
+                pthread_rwlock_unlock(&poas.lock);
+                return -EBUSY;
+        }
+
+        list_del(&found->next);
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        poa_teardown(found);
+
+        return 0;
+}
+
+static int poa_del(const struct poa_spec * spec)
+{
+        struct poa_detach_req req;
+        int                   cs;
+        int                   ret;
+
+        memset(&req, 0, sizeof(req));
+
+        req.spec = *spec;
+
+        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cs);
+
+        if (poas.state < POA_RUNNING) { /* set before workers run */
+                ret = poa_do_detach(&req);
+                goto out;
+        }
+
+        pthread_mutex_lock(&poas.mgmt_mtx);
+
+        if (poas.mgmt_stop) { /* stopped: poa_fini reaps these */
+                pthread_mutex_unlock(&poas.mgmt_mtx);
+
+                ret = -EBUSY;
+                goto out;
+        }
+
+        list_add_tail(&req.next, &poas.detach);
+
+        pthread_cond_signal(&poas.mgmt_cond);
+
+        while (!req.done)
+                pthread_cond_wait(&poas.done_cond, &poas.mgmt_mtx);
+
+        pthread_mutex_unlock(&poas.mgmt_mtx);
+
+        ret = req.result;
+ out:
+        pthread_setcancelstate(cs, NULL);
+
+        return ret;
+}
+
+static __inline__ bool mgmt_idle(void)
+{
+        if (poas.mgmt_stop)
+                return false;
+
+        if (!llist_is_empty(&poas.mgmt_frames))
+                return false;
+
+        return list_is_empty(&poas.detach);
+}
+
+static void detach_run(void)
+{
+        while (!list_is_empty(&poas.detach)) {
+                struct poa_detach_req * req;
+
+                req = list_first_entry(&poas.detach, struct poa_detach_req,
+                                       next);
+                list_del(&req->next);
+
+                pthread_mutex_unlock(&poas.mgmt_mtx);
+
+                req->result = poa_do_detach(req);
+
+                pthread_mutex_lock(&poas.mgmt_mtx);
+
+                req->done = true;
+
+                pthread_cond_broadcast(&poas.done_cond);
+        }
+}
+
+static void * mgmt_handler(void * o)
+{
+        struct timespec intv = TIMESPEC_INIT_MS(POA_SWEEP_TIMEO);
+
+        (void) o;
+
+        while (true) {
+                struct poa_mgmt_frame * frame;
+                struct timespec         abstime;
+
+                pthread_mutex_lock(&poas.mgmt_mtx);
+
+                detach_run();
+
+                while (mgmt_idle()) {
+                        clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+                        ts_add(&abstime, &intv, &abstime);
+
+                        if (pthread_cond_timedwait(&poas.mgmt_cond,
+                                                   &poas.mgmt_mtx,
+                                                   &abstime) == ETIMEDOUT) {
+                                pthread_mutex_unlock(&poas.mgmt_mtx);
+                                sweep_pending();
+                                pthread_mutex_lock(&poas.mgmt_mtx);
+                        }
+                }
+
+                if (poas.mgmt_stop) {
+                        detach_run(); /* nobody else serves these */
+
+                        pthread_mutex_unlock(&poas.mgmt_mtx);
+                        break;
+                }
+
+                if (llist_is_empty(&poas.mgmt_frames)) {
+                        pthread_mutex_unlock(&poas.mgmt_mtx);
+                        continue;
+                }
+
+                frame = llist_first_entry(&poas.mgmt_frames,
+                                          struct poa_mgmt_frame, next);
+                llist_del(&frame->next, &poas.mgmt_frames);
+
+                --frame->poa->n_mgmt;
+
+                pthread_mutex_unlock(&poas.mgmt_mtx);
+
+                mgmt_frame_handle(frame);
+
+                free(frame);
+        }
+
+        return (void *) 0;
+}
+
+#ifndef HAVE_ETH
+/* Only the Ethernet transport reports link events. */
+int poa_monitor_open(void)
+{
+        return -1;
+}
+
+void poa_monitor_read(int fd)
+{
+        (void) fd;
+}
+#endif
+
+static void * poa_monitor(void * o)
+{
+        (void) o;
+
+        while (true)
+                poa_monitor_read(poas.mon_fd);
+
+        return (void *) 0;
+}
+
+static int monitor_start(void)
+{
+        poas.mon_fd = poa_monitor_open();
+        if (poas.mon_fd < 0)
+                return 0;
+
+        if (pthread_create(&poas.monitor, NULL, poa_monitor, NULL) == 0)
+                return 0;
+
+        close(poas.mon_fd);
+
+        poas.mon_fd = -1;
+
+        return -1;
+}
+
+static void monitor_stop(void)
+{
+        if (poas.mon_fd < 0)
+                return;
+
+        pthread_cancel(poas.monitor);
+        pthread_join(poas.monitor, NULL);
+
+        close(poas.mon_fd);
+
+        poas.mon_fd = -1;
+}
+
+int poa_start(void)
+{
+        struct list_head * p;
+        struct list_head * h;
+
+        if (poas.state == POA_NULL)
+                return 0;
+
+        if (pthread_create(&poas.mgmt_handler, NULL, mgmt_handler, NULL) != 0)
+                goto fail_mgmt;
+
+        if (monitor_start() < 0)
+                goto fail_monitor;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->ops->poa_start(poa) < 0)
+                        goto fail_reader;
+        }
+
+        poas.state = POA_RUNNING;
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return 0;
+
+ fail_reader:
+        list_for_each(h, &poas.list) {
+                struct poa * poa = list_entry(h, struct poa, next);
+
+                if (h == p)
+                        break;
+
+                poa->ops->poa_stop(poa);
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        monitor_stop();
+ fail_monitor:
+        pthread_mutex_lock(&poas.mgmt_mtx);
+
+        poas.mgmt_stop = true;
+
+        pthread_cond_broadcast(&poas.mgmt_cond);
+        pthread_mutex_unlock(&poas.mgmt_mtx);
+
+        pthread_join(poas.mgmt_handler, NULL);
+ fail_mgmt:
+        return -1;
+}
+
+void poa_stop(void)
+{
+        struct list_head * p;
+
+        if (poas.state < POA_RUNNING)
+                return;
+
+        pthread_mutex_lock(&poas.mgmt_mtx);
+
+        poas.mgmt_stop = true;
+
+        pthread_cond_broadcast(&poas.mgmt_cond);
+        pthread_mutex_unlock(&poas.mgmt_mtx);
+
+        pthread_join(poas.mgmt_handler, NULL);
+
+        monitor_stop();
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                poa->ops->poa_stop(poa);
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        poas.state = POA_INIT;
+}
+
+void poa_fini(void)
+{
+        if (poas.state == POA_NULL)
+                return;
+
+        poa_stop();
+
+        poas.state = POA_NULL;
+
+        poa_detach_all();
+
+#ifdef PROC_FLOW_STATS
+        rib_unreg(POA_RIB);
+#endif
+
+        pthread_rwlock_destroy(&poas.lock);
+        pthread_cond_destroy(&poas.done_cond);
+        pthread_cond_destroy(&poas.mgmt_cond);
+        pthread_mutex_destroy(&poas.mgmt_mtx);
+        pthread_cond_destroy(&poas.cond);
+        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)
+{
+        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;
+        }
+
+        for (i = 0; i < POA_MAX_POAS; i++)
+                if (!used[i])
+                        return i;
+
+        return -1;
+}
+
+static int poa_add(const struct poa_spec * spec,
+                   const struct poa_ops *  ops,
+                   size_t                  n_eids)
+{
+        struct list_head * p;
+        struct poa *       poa;
+        int                err;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                poa = list_entry(p, struct poa, next);
+                if (poa_has_id(poa, spec)) {
+                        pthread_rwlock_unlock(&poas.lock);
+                        return -EPERM;
+                }
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        poa = poa_create(spec->type, ops, n_eids);
+        if (poa == NULL)
+                return -ENOMEM;
+
+        err = poa->ops->poa_attach(poa, spec);
+        if (err < 0)
+                goto fail_bind;
+
+        err = poa_addr_name(&poa->local, poa->name, sizeof(poa->name));
+        if (err < 0)
+                goto fail_start;
+
+        err = -1;
+
+        pthread_rwlock_wrlock(&poas.lock);
+
+        poa->qid = poa_qid_alloc();
+        if (poa->qid < 0) {
+                pthread_rwlock_unlock(&poas.lock);
+                goto fail_start;
+        }
+
+        if (poas.state >= POA_RUNNING && poa->ops->poa_start(poa) < 0) {
+                pthread_rwlock_unlock(&poas.lock);
+                goto fail_start;
+        }
+
+        list_add_tail(&poa->next, &poas.list);
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return 0;
+
+ fail_start:
+        poa->ops->poa_detach(poa);
+ fail_bind:
+        poa_destroy(poa);
+        return err;
+}
+
+/* The single place a type is bound to its transport. */
+int poa_attach(const struct poa_spec * poa)
+{
+        if (poa == NULL)
+                return -EINVAL;
+
+        switch (poa->type) {
+        case POA_UDP4:
+                /* FALLTHRU */
+        case POA_UDP6:
+                return poa_add(poa, &udp_poa_ops, POA_MAX_EIDS);
+        case POA_ETH:
+#ifdef HAVE_ETH
+                return poa_add(poa, &eth_poa_ops, POA_MAX_EIDS);
+#else
+                return -ENOTSUP;
+#endif
+        default:
+                return -ENOTSUP;
+        }
+}
+
+int poa_detach(const struct poa_spec * poa)
+{
+        if (poa == NULL)
+                return -EINVAL;
+
+        return poa_del(poa);
+}
+
+ssize_t poa_list(struct poa_spec * specs,
+                 size_t            max)
+{
+        struct list_head * p;
+        size_t             n = 0;
+
+        if (specs == NULL)
+                return -EINVAL;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (n++ >= max)
+                        continue;
+
+                memset(specs, 0, sizeof(*specs));
+
+                poa->ops->poa_spec(poa, specs);
+
+                specs++;
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        return (ssize_t) n;
+}
+
+/*
+ * Complete peer for dst on the backend serving its type. The ops are
+ * borrowed under poas.lock and called outside it (see poa_query); if
+ * every PoA of the type detaches in between, the query's broadcast
+ * reaches nothing and reports -EPERM, as the lookup would.
+ */
+static int poa_peer_resolve(const char *      dst,
+                            struct poa_addr * peer)
+{
+        const struct poa_ops * ops = NULL;
+        struct list_head *     p;
+
+        pthread_rwlock_rdlock(&poas.lock);
+
+        list_for_each(p, &poas.list) {
+                struct poa * poa = list_entry(p, struct poa, next);
+
+                if (poa->type == peer->type) {
+                        ops = poa->ops;
+                        break;
+                }
+        }
+
+        pthread_rwlock_unlock(&poas.lock);
+
+        if (ops == NULL) /* nothing could carry the flow */
+                return -EPERM;
+
+        if (ops->poa_query == NULL) /* these addresses arrive complete */
+                return 0;
+
+        return ops->poa_query(dst, NULL, peer);
+}
+
+/*
+ * Three steps: the IRMd creates the flow and prepares the key exchange,
+ * the PoA handshakes with the peer, the IRMd completes the
+ * exchange and hands us the key.
+ */
+int poa_flow_alloc(const char *            dst,
+                   const struct poa_addr * addr,
+                   qosspec_t *             qs,
+                   const struct timespec * timeo)
+{
+        struct flow_info  flow;
+        struct poa_flow * pf;
+        struct poa_addr   peer;
+        struct crypt_sk   crypt;
+        struct timespec   t0;
+        struct timespec   t1;
+        uint8_t           key[SYMMKEYSZ];
+        uint8_t           buf[SOCK_BUF_SIZE];
+        buffer_t          msg = {SOCK_BUF_SIZE, buf};
+        buffer_t          req;
+        buffer_t          resp;
+        uint32_t          mtu = 0;
+        int               err;
+
+        if (addr == NULL)
+                return -EINVAL;
+
+        if (qs != NULL && qs->service == SVC_STREAM && qs->loss != 0)
+                return -EINVAL;
+
+        peer = *addr;
+
+        err = poa_peer_resolve(dst, &peer);
+        if (err < 0)
+                return err;
+
+        addr = &peer;
+
+        err = poa_check(addr);
+        if (err < 0)
+                return err;
+
+        memset(&flow, 0, sizeof(flow));
+
+        flow.n_pid   = getpid();
+        flow.n_1_pid = getpid();
+        flow.qs      = qs == NULL ? qos_raw : *qs;
+        if (poa_flow_alloc__irm_req_ser(&msg, &flow, dst) < 0)
+                return -ENOMEM;
+
+        err = send_recv_msg(&msg);
+        if (err < 0)
+                return err;
+
+        clrbuf(req);
+        clrbuf(resp);
+
+        err = poa_flow__irm_result_des(&msg, &flow, &req);
+        if (err < 0)
+                return err;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &t0);
+
+        err = poa_alloc(addr, flow.qs, &req, &resp, &pf, &mtu, timeo);
+
+        freebuf(req);
+
+        if (err < 0)
+                goto fail_alloc;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &t1);
+
+        flow.mtu = mtu;
+        flow.mpl = pf->poa->mpl;
+        msg.len  = SOCK_BUF_SIZE;
+        msg.data = buf;
+        if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, &resp, 0) < 0) {
+                err = -ENOMEM;
+                goto fail_resp;
+        }
+
+        freebuf(resp);
+
+        err = send_recv_msg(&msg);
+        if (err < 0)
+                goto fail_msg;
+
+        crypt.key   = key;
+        crypt.epoch = 0;
+        crypt.role  = CRYPT_ROLE_INIT;
+
+        err = flow__irm_result_des(&msg, &flow, &crypt);
+        if (err < 0)
+                goto fail_msg;
+
+        err = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), pf);
+
+        crypt_secure_clear(key, SYMMKEYSZ);
+
+        if (err < 0)
+                goto fail_msg;
+
+        if (qs != NULL)
+                *qs = flow.qs;
+
+        return err;
+
+ fail_resp:
+        freebuf(resp);
+ fail_msg:
+        poa_alloc_fail(pf);
+        return err;
+ fail_alloc:
+        msg.len  = SOCK_BUF_SIZE;
+        msg.data = buf;
+        if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, NULL, err) == 0)
+                send_recv_msg(&msg);
+
+        return err;
+}
+
diff --git a/src/lib/poa/poa.h b/src/lib/poa/poa.h
new file mode 100644
index 00000000..014986a3
--- /dev/null
+++ b/src/lib/poa/poa.h
@@ -0,0 +1,364 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - internal API
+ *
+ *    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_POA_POA_H
+#define OUROBOROS_LIB_POA_POA_H
+
+#include <ouroboros/atomics.h>
+#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/list.h>
+#include <ouroboros/qos.h>
+#include <ouroboros/rcu.h>
+#include <ouroboros/ssm_flow_set.h>
+#include <ouroboros/ssm_pool.h>
+#include <ouroboros/ssm_rbuff.h>
+#include <ouroboros/time.h>
+#include <ouroboros/utils.h>
+
+#include <errno.h>
+#include <limits.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+#define POA_MGMT_EID     0  /* reserved for the mgmt channel */
+
+#define POA_FLOW_REQ     1
+#define POA_FLOW_REPLY   2
+#define POA_FLOW_UPDATE  3
+#define POA_NAME_QUERY   4
+#define POA_NAME_REPLY   5
+
+#define POA_QUERY_HLEN   32 /* SHA3-256, the query hash algorithm */
+
+/* Fits "udp6.<ip6>.<port>", the longest display/RIB entry name. */
+#define POA_NAME_STRLEN  63
+
+/* Management message; every transport frames it the same way. */
+struct poa_mgmt_msg {
+        uint8_t  code;
+        uint8_t  resv[3];
+        uint32_t s_eid;
+        uint32_t d_eid;
+        int32_t  response;
+        uint64_t bandwidth;
+        uint32_t delay;
+        uint32_t loss;
+        uint32_t ber;
+        uint32_t max_gap;
+        uint32_t timeout;
+        uint8_t  availability;
+        uint8_t  service;
+        uint16_t data_len;
+} __attribute__((packed));
+
+struct poa;
+struct poa_flow;
+
+#ifdef PROC_FLOW_STATS
+#define POA_STAT_BUMP(poa, field)   FETCH_ADD_RELAXED(&(poa)->stat.field, 1)
+#define POA_STAT_ADD(poa, field, v) FETCH_ADD_RELAXED(&(poa)->stat.field, (v))
+#define POA_STAT_SUB(poa, field, v) FETCH_SUB_RELAXED(&(poa)->stat.field, (v))
+#define POA_STAT_LOAD(poa, field)   LOAD_RELAXED(&(poa)->stat.field)
+#else
+#define POA_STAT_BUMP(poa, field)   ((void) (poa))
+#define POA_STAT_ADD(poa, field, v) ((void) (poa), (void) (v))
+#define POA_STAT_SUB(poa, field, v) ((void) (poa), (void) (v))
+#define POA_STAT_LOAD(poa, field)   ((void) (poa), (size_t) 0)
+#endif
+
+struct poa_stat {
+        size_t n_flows;  /* gauge; the RIB reads it without poas.lock */
+        size_t rx_pkts;  /* packets delivered to a flow               */
+        size_t rx_bytes; /* payload bytes delivered                   */
+        size_t tx_pkts;  /* packets handed to the transport           */
+        size_t tx_bytes; /* payload bytes handed to the transport     */
+        size_t mgmt_rx;  /* frames queued for the mgmt handler        */
+        size_t mgmt_tx;  /* management frames sent                    */
+        size_t bad_eid;  /* no flow on the EID a peer sent            */
+        size_t dlv_fail; /* the rx ring above would not take it       */
+        size_t buf_fail; /* no buffer to receive into                 */
+        size_t rcv_fail; /* transport read failed; the reader exits   */
+        size_t snd_fail; /* transport send failed                     */
+        size_t qry_tx;   /* name queries broadcast                    */
+        size_t qry_rx;   /* name queries received                     */
+        size_t rep_tx;   /* name replies sent, query matched          */
+        size_t rep_rx;   /* name replies received                     */
+};
+
+/* Spacing between transmit-depth samples; a depth costs a syscall. */
+#define POA_QLEN_GATE   (100 * 1000) /* ns */
+
+
+/* Transport operations; public poa_X() dispatches to ops->poa_X. */
+struct poa_ops {
+        /* Parse own arm of the spec; validate; fill local and priv. */
+        int      (* poa_attach)(struct poa *            poa,
+                                const struct poa_spec * spec);
+
+        void     (* poa_detach)(struct poa * poa);
+
+        int      (* poa_start)(struct poa * poa);
+
+        void     (* poa_stop)(struct poa * poa);
+
+        /* Full queue: -EAGAIN unless block; then wait, to abstime if set. */
+        int      (* poa_send)(struct poa *            poa,
+                              const struct poa_addr * dst,
+                              uint32_t                eid,
+                              struct ssm_pk_buff *    spb,
+                              bool                    block,
+                              const struct timespec * abstime);
+
+        int      (* poa_send_mgmt)(struct poa *            poa,
+                                   const struct poa_addr * dst,
+                                   const uint8_t *         buf,
+                                   size_t                  len);
+
+        int      (* poa_query)(const char *            dst,
+                               const struct timespec * timeo,
+                               struct poa_addr *       addr);
+
+        uint32_t (* poa_mtu)(struct poa *            poa,
+                             const struct poa_addr * dst);
+
+        /* Bytes queued in the transmit path of the PoA. */
+        size_t   (* poa_qlen)(struct poa * poa);
+
+        /* Depth from the queue itself; NULL infers it from qlen. */
+        int      (* poa_qpkts)(struct poa * poa,
+                               size_t *     pkts,
+                               size_t *     byts);
+
+        int      (* poa_rib)(struct poa * poa,
+                             char *       buf,
+                             size_t       len);
+
+        /* Identity as a spec, e.g. for poa_list. */
+        void     (* poa_spec)(const struct poa * poa,
+                              struct poa_spec *  spec);
+
+        /* Same identity as spec? Caller matched poa->type already. */
+        bool     (* poa_has_id)(const struct poa *      poa,
+                                const struct poa_spec * spec);
+
+        /* Carries dst? Caller matched poa->type already. */
+        bool     (* poa_match)(const struct poa *      poa,
+                               const struct poa_addr * dst);
+
+        /*
+         * Flows ride the link this id names; NULL: no link events.
+         * Ids are meaningful only to the backend whose monitor
+         * produced them; a single backend owns the monitor.
+         */
+        bool     (* poa_link_match)(const struct poa * poa,
+                                    int                id);
+
+        /* Maximum packet lifetime in the transport, seconds. */
+        time_t   mpl;
+};
+struct poa {
+        struct list_head       next;
+
+        enum poa_type          type;
+        const struct poa_ops * ops;
+        void *                 priv;
+
+        struct poa_addr        local;   /* what peers dial us on */
+
+        /* Display/RIB entry name for local, e.g. "udp4.<ip>.<port>". */
+        char                   name[POA_NAME_STRLEN + 1];
+
+        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. */
+        size_t                 avg_ovh;
+        size_t                 n_tx;
+
+        /* Last queue depth read, and when, in the transport's terms. */
+        size_t                 q_cache;
+        uint64_t               q_time;
+
+        /* Queued management frames, capped; poas.mgmt_mtx guards. */
+        size_t                 n_mgmt;
+
+#ifdef PROC_FLOW_STATS
+        struct poa_stat        stat;
+#endif
+
+        struct bmp *           eids;
+        struct poa_flow **     eid_to_pf;
+        size_t                 n_eids;
+
+        struct list_head       flows;     /* live flows, for repeats */
+
+        /* Keeps a flow and its ring alive under the receive path. */
+        struct rcu_guard       guard;
+};
+
+/* poa/poa.c is part of the dev.c translation unit. */
+int               poa_init(const char * name);
+
+int               poa_start(void);
+
+void              poa_stop(void);
+
+void              poa_fini(void);
+
+/* Also answer name queries for the layer once enrolled. */
+int               poa_set_layer(const char * layer);
+
+/* Hash of a name this process answers queries for? */
+bool              poa_has_name(const uint8_t * hash);
+
+int               poa_flow_tx(struct poa_flow *       pf,
+                              struct ssm_pk_buff *    spb,
+                              bool                    block,
+                              const struct timespec * abstime);
+
+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);
+
+size_t            poa_flow_mean_len(const struct poa_flow * pf);
+
+void              poa_flow_attach(struct poa_flow *  pf,
+                                  int                flow_id,
+                                  struct ssm_rbuff * rx_rb);
+
+void              poa_flow_ready(struct poa_flow * pf);
+
+void              poa_flow_detach(struct poa_flow * pf);
+
+struct poa_flow * poa_flow_take_pending(int flow_id);
+
+/* Addresses and management messages (poa/addr.c). */
+int               poa_addr_cmp(const struct poa_addr * a,
+                               const struct poa_addr * b);
+
+/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */
+int               poa_addr_name(const struct poa_addr * a,
+                                char *                  buf,
+                                size_t                  len);
+
+void              poa_mgmt_msg_ser(struct poa_mgmt_msg * msg,
+                                   uint8_t               code,
+                                   uint32_t              s_eid,
+                                   uint32_t              d_eid,
+                                   qosspec_t             qs,
+                                   int                   response,
+                                   size_t                data_len);
+
+void              poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg,
+                                   qosspec_t *                 qs);
+
+/* Called by the transports. */
+void              poa_rx_pkt(struct poa *         poa,
+                             uint32_t             eid,
+                             struct ssm_pk_buff * spb);
+
+void              poa_rx_mgmt(struct poa *            poa,
+                              const struct poa_addr * src,
+                              const uint8_t *         buf,
+                              size_t                  len);
+
+/* Reserve a buffer for a received packet, with transport headroom. */
+int               poa_spb_reserve(struct ssm_pk_buff ** spb,
+                                  size_t                len);
+
+void              poa_spb_release(struct ssm_pk_buff * spb);
+
+/*
+ * All flows on PoAs whose poa_link_match reports this link id go up
+ * or down with it. Returns the number of flows whose state changed.
+ */
+size_t            poa_link_updown(int  id,
+                                  bool up);
+
+/*
+ * Link monitor: one socket for the whole subsystem, opened by
+ * poa_start(). Returns -1 where the transport has no monitor.
+ */
+int               poa_monitor_open(void);
+
+/* Reads one batch of link events; cancellation point. */
+void              poa_monitor_read(int fd);
+
+/* Broadcast a mgmt frame on every PoA matching dst; # sent. */
+int               poa_bcast_mgmt(const struct poa_addr * dst,
+                                 const uint8_t *         buf,
+                                 size_t                  len);
+
+/* Transport op tables. */
+extern const struct poa_ops udp_poa_ops;
+extern const struct poa_ops eth_poa_ops;
+
+/*
+ * Waits for a descriptor to take another packet, up to abstime.
+ * A NULL deadline waits indefinitely. Transports call this when
+ * their send reports the transmit queue full.
+ */
+static __inline__ int poa_wait_out(int                     fd,
+                                   const struct timespec * abstime)
+{
+        struct pollfd   pfd;
+        struct timespec now;
+        long            ms      = -1;
+        bool            clamped = false;
+        int             ret;
+
+        if (abstime != NULL) {
+                clock_gettime(PTHREAD_COND_CLOCK, &now);
+
+                if (ts_diff_ns(abstime, &now) <= 0)
+                        return -ETIMEDOUT;
+
+                ms = ts_diff_ms(abstime, &now) + 1; /* sub-ms must wait */
+                if (ms > INT_MAX) { /* poll takes an int */
+                        ms      = INT_MAX;
+                        clamped = true;
+                }
+        }
+
+        pfd.fd      = fd;
+        pfd.events  = POLLOUT;
+        pfd.revents = 0;
+
+        ret = poll(&pfd, 1, (int) ms);
+        if (ret < 0)
+                return errno == EINTR ? 0 : -EIO;
+
+        if (ret == 0)
+                return clamped ? 0 : -ETIMEDOUT; /* clamped: retry */
+
+        return 0;
+}
+
+#endif /* OUROBOROS_LIB_POA_POA_H */
diff --git a/src/lib/poa/udp.c b/src/lib/poa/udp.c
new file mode 100644
index 00000000..6753347a
--- /dev/null
+++ b/src/lib/poa/udp.c
@@ -0,0 +1,633 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Points of attachment (PoA) - UDP transport
+ *
+ *    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/.
+ */
+
+#if defined(__APPLE__)
+#define _BSD_SOURCE
+#define _DARWIN_C_SOURCE
+#elif defined(__FreeBSD__)
+#define __BSD_VISIBLE 1
+#elif 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"
+
+#define OUROBOROS_PREFIX "poa-udp"
+
+#include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/logs.h>
+#include <ouroboros/time.h>
+#include <ouroboros/utils.h>
+
+#include "poa.h"
+
+#ifdef __linux__
+#include <linux/sockios.h>
+#endif
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define UDP_HDR_LEN     sizeof(uint32_t) /* PoA id */
+#define UDP_MAX_PAYLOAD (POA_UDP_RD_BUF - UDP_HDR_LEN)
+/* The reader buffer must fit a full mgmt frame at any tuning. */
+#define UDP_MAX_PACKET  MAX(POA_UDP_RD_BUF, POA_MGMT_FRAME_SIZE + UDP_HDR_LEN)
+#define UDP_IP4_OVERH   28U              /* IPv4 + UDP */
+#define UDP_IP6_OVERH   48U              /* IPv6 + UDP */
+/* Wait for the link to come back before reading it again. */
+#define UDP_DOWN_TIMEO  100              /* ms */
+
+union udp_saddr {
+        struct sockaddr     sa;
+        struct sockaddr_in  in;
+        struct sockaddr_in6 in6;
+};
+
+struct udp_priv {
+        int             s_fd;
+        int             af;
+        union udp_saddr s_saddr;
+        pthread_t       reader;
+        bool            running;
+};
+
+static socklen_t saddr_len(int af)
+{
+        if (af == AF_INET)
+                return sizeof(struct sockaddr_in);
+
+        return sizeof(struct sockaddr_in6);
+}
+
+static void addr_to_saddr(const struct poa_addr * addr,
+                          union udp_saddr *       saddr)
+{
+        memset(saddr, 0, sizeof(*saddr));
+
+        if (addr->type == POA_UDP4) {
+                saddr->in.sin_family = AF_INET;
+                saddr->in.sin_addr   = addr->udp4.ip_addr;
+                saddr->in.sin_port   = htons(addr->udp4.port);
+        } else {
+                saddr->in6.sin6_family = AF_INET6;
+                saddr->in6.sin6_addr   = addr->udp6.ip_addr;
+                saddr->in6.sin6_port   = htons(addr->udp6.port);
+        }
+}
+
+static void saddr_to_addr(const union udp_saddr * saddr,
+                          struct poa_addr *       addr)
+{
+        memset(addr, 0, sizeof(*addr));
+
+        if (saddr->sa.sa_family == AF_INET) {
+                addr->type         = POA_UDP4;
+                addr->udp4.ip_addr = saddr->in.sin_addr;
+                addr->udp4.port    = ntohs(saddr->in.sin_port);
+        } else {
+                addr->type         = POA_UDP6;
+                addr->udp6.ip_addr = saddr->in6.sin6_addr;
+                addr->udp6.port    = ntohs(saddr->in6.sin6_port);
+        }
+}
+
+/* A datagram longer than the buffer arrives truncated: drop it. */
+static void * udp_reader(void * o)
+{
+        struct poa *      poa  = (struct poa *) o;
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+        struct timespec   down = TIMESPEC_INIT_MS(UDP_DOWN_TIMEO);
+        uint8_t *         buf;
+
+        buf = malloc(UDP_MAX_PACKET);
+        if (buf == NULL)
+                return (void *) -1;
+
+        pthread_cleanup_push(free, buf);
+
+        while (true) {
+                struct ssm_pk_buff * spb;
+                union udp_saddr      r_saddr;
+                struct poa_addr      src;
+                struct msghdr        mh;
+                struct iovec         iov;
+                ssize_t              n;
+                uint32_t             eid;
+                size_t               plen;
+
+                iov.iov_base = buf;
+                iov.iov_len  = UDP_MAX_PACKET;
+
+                memset(&mh, 0, sizeof(mh));
+
+                mh.msg_name    = &r_saddr;
+                mh.msg_namelen = sizeof(r_saddr);
+                mh.msg_iov     = &iov;
+                mh.msg_iovlen  = 1;
+
+                n = recvmsg(priv->s_fd, &mh, 0);
+                if (n < 0) {
+                        if (errno == EINTR)
+                                continue;
+
+                        POA_STAT_BUMP(poa, rcv_fail);
+
+                        if (errno == ENETDOWN) {
+                                nanosleep(&down, NULL);
+                                continue;
+                        }
+
+                        log_err("Reader stopped: %s.", strerror(errno));
+                        break;
+                }
+
+                if ((mh.msg_flags & MSG_TRUNC) != 0) {
+                        POA_STAT_BUMP(poa, buf_fail);
+                        continue;
+                }
+
+                if ((size_t) n < UDP_HDR_LEN)
+                        continue;
+
+                eid  = ntoh32(*(uint32_t *) buf);
+                plen = (size_t) n - UDP_HDR_LEN;
+
+                saddr_to_addr(&r_saddr, &src);
+
+                if (eid == POA_MGMT_EID) {
+                        poa_rx_mgmt(poa, &src, buf + UDP_HDR_LEN, plen);
+                        continue;
+                }
+
+                if (poa_spb_reserve(&spb, plen) < 0) {
+                        POA_STAT_BUMP(poa, buf_fail);
+                        continue;
+                }
+
+                memcpy(ssm_pk_buff_head(spb), buf + UDP_HDR_LEN, plen);
+
+                poa_rx_pkt(poa, eid, spb);
+        }
+
+        pthread_cleanup_pop(true);
+
+        return (void *) 0;
+}
+
+/* Reads the bound address back: an ephemeral port is only known after. */
+static int udp_attach(struct poa *            poa,
+                      const struct poa_spec * spec)
+{
+        struct udp_priv * priv;
+        socklen_t         len;
+        int               af;
+
+        af = spec->type == POA_UDP4 ? AF_INET : AF_INET6;
+
+        priv = malloc(sizeof(*priv));
+        if (priv == NULL)
+                return -ENOMEM;
+
+        memset(priv, 0, sizeof(*priv));
+
+        priv->af   = af;
+        priv->s_fd = socket(af, SOCK_DGRAM, IPPROTO_UDP);
+        if (priv->s_fd < 0) {
+                log_err("Failed to create socket: %s.", strerror(errno));
+                goto fail_socket;
+        }
+
+        if (af == AF_INET) {
+                priv->s_saddr.in.sin_family = AF_INET;
+                priv->s_saddr.in.sin_addr   = spec->udp4.ip_addr;
+                priv->s_saddr.in.sin_port   = htons(spec->udp4.port);
+        } else {
+                int on = 1;
+
+                if (setsockopt(priv->s_fd, IPPROTO_IPV6, IPV6_V6ONLY,
+                               &on, sizeof(on)) < 0) {
+                        log_err("Failed to set IPV6_V6ONLY: %s.",
+                                strerror(errno));
+                        goto fail_bind;
+                }
+
+                priv->s_saddr.in6.sin6_family = AF_INET6;
+                priv->s_saddr.in6.sin6_addr   = spec->udp6.ip_addr;
+                priv->s_saddr.in6.sin6_port   = htons(spec->udp6.port);
+        }
+
+        if (bind(priv->s_fd, &priv->s_saddr.sa, saddr_len(af)) < 0) {
+                log_err("Failed to bind: %s.", strerror(errno));
+                goto fail_bind;
+        }
+
+        poa->priv = priv;
+        len       = saddr_len(af);
+        if (getsockname(priv->s_fd, &priv->s_saddr.sa, &len) < 0)
+                log_warn("Failed to read the bound address: %s.",
+                         strerror(errno));
+
+        saddr_to_addr(&priv->s_saddr, &poa->local);
+
+        return 0;
+
+ fail_bind:
+        close(priv->s_fd);
+ fail_socket:
+        poa->priv = NULL;
+
+        free(priv);
+
+        return -EIO;
+}
+
+static void udp_detach(struct poa * poa)
+{
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+        if (priv == NULL)
+                return;
+
+        close(priv->s_fd);
+
+        free(priv);
+
+        poa->priv = NULL;
+}
+
+/* One reader per socket, so a flow cannot be reordered on receive. */
+static int udp_start(struct poa * poa)
+{
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+        if (pthread_create(&priv->reader, NULL, udp_reader, poa) != 0)
+                return -1;
+
+        priv->running = true;
+
+        return 0;
+}
+
+static void udp_stop(struct poa * poa)
+{
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+
+        if (!priv->running)
+                return;
+
+        pthread_cancel(priv->reader);
+        pthread_join(priv->reader, NULL);
+
+        priv->running = false;
+}
+
+/*
+ * The PoA id is a separate iovec, so the buffer needs no headroom.
+ * MSG_DONTWAIT: the reader blocks on this socket.
+ */
+static int udp_sendv(struct udp_priv *       priv,
+                     const struct poa_addr * dst,
+                     uint32_t                eid,
+                     const uint8_t *         body,
+                     size_t                  len,
+                     bool                    block,
+                     const struct timespec * abstime)
+{
+        union udp_saddr saddr;
+        struct msghdr   msg;
+        struct iovec    iov[2];
+        uint32_t        hdr;
+        int             ret;
+
+        if (len > UDP_MAX_PAYLOAD)
+                return -EMSGSIZE;
+
+        addr_to_saddr(dst, &saddr);
+
+        hdr = hton32(eid);
+
+        iov[0].iov_base = &hdr;
+        iov[0].iov_len  = sizeof(hdr);
+        iov[1].iov_base = (void *) body;
+        iov[1].iov_len  = len;
+
+        memset(&msg, 0, sizeof(msg));
+
+        msg.msg_name    = &saddr;
+        msg.msg_namelen = saddr_len(priv->af);
+        msg.msg_iov     = iov;
+        msg.msg_iovlen  = len > 0 ? 2 : 1;
+        while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) {
+                if (errno != EAGAIN && errno != EWOULDBLOCK)
+                        return -EIO;
+
+                if (!block)
+                        return -EAGAIN;
+
+                ret = poa_wait_out(priv->s_fd, abstime);
+                if (ret < 0)
+                        return ret;
+        }
+
+        return 0;
+}
+
+static int udp_send(struct poa *            poa,
+                    const struct poa_addr * dst,
+                    uint32_t                eid,
+                    struct ssm_pk_buff *    spb,
+                    bool                    block,
+                    const struct timespec * abstime)
+{
+        return udp_sendv((struct udp_priv *) poa->priv, dst, eid,
+                         ssm_pk_buff_head(spb), ssm_pk_buff_len(spb),
+                         block, abstime);
+}
+
+static int udp_send_mgmt(struct poa *            poa,
+                         const struct poa_addr * dst,
+                         const uint8_t *         buf,
+                         size_t                  len)
+{
+        struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO);
+        struct timespec abstime;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+        ts_add(&abstime, &timeo, &abstime);
+
+        return udp_sendv((struct udp_priv *) poa->priv, dst, POA_MGMT_EID,
+                         buf, len, true, &abstime);
+}
+
+/* The PoA id header eats into the usable MTU. */
+static uint32_t udp_mtu(struct poa *            poa,
+                        const struct poa_addr * dst)
+{
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+        uint32_t          fallback;
+        uint32_t          overh;
+#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU))
+        union udp_saddr   saddr;
+        socklen_t         len;
+        int               sock;
+        int               mtu = 0;
+#endif
+        if (priv->af == AF_INET) {
+                fallback = POA_UDP4_MTU;
+                overh    = UDP_IP4_OVERH;
+        } else {
+                fallback = POA_UDP6_MTU;
+                overh    = UDP_IP6_OVERH;
+        }
+
+        fallback -= UDP_HDR_LEN;
+        if (fallback > UDP_MAX_PAYLOAD)
+                fallback = UDP_MAX_PAYLOAD;
+
+#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU))
+
+        addr_to_saddr(dst, &saddr);
+
+        sock = socket(priv->af, SOCK_DGRAM, IPPROTO_UDP);
+        if (sock < 0)
+                return fallback;
+
+        if (connect(sock, &saddr.sa, saddr_len(priv->af)) < 0)
+                goto fallback;
+
+        len = sizeof(mtu);
+
+#if defined(IP_MTU)
+        if (priv->af == AF_INET) {
+                if (getsockopt(sock, IPPROTO_IP, IP_MTU, &mtu, &len) < 0)
+                        goto fallback;
+        }
+#endif
+#if defined(IPV6_MTU)
+        if (priv->af == AF_INET6) {
+                if (getsockopt(sock, IPPROTO_IPV6, IPV6_MTU, &mtu, &len) < 0)
+                        goto fallback;
+        }
+#endif
+        close(sock);
+
+        if (mtu <= (int) (overh + UDP_HDR_LEN))
+                return fallback;
+
+        return MIN((uint32_t) mtu - overh - UDP_HDR_LEN, UDP_MAX_PAYLOAD);
+
+ fallback:
+        close(sock);
+#else
+        (void) dst;
+        (void) overh;
+#endif
+        return fallback;
+}
+
+/* All flows on the PoA share the socket, so this is aggregate. */
+static size_t udp_qlen(struct poa * poa)
+{
+#if defined(__linux__) && defined(SIOCOUTQ)
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+        int               qlen;
+
+        qlen = 0;
+        if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0)
+                return 0;
+
+        return (size_t) qlen;
+#else
+        (void) poa;
+
+        return 0;
+#endif
+}
+
+/* The kernel keeps no per-socket drop count for UDP. */
+static int udp_rib(struct poa * poa,
+                   char *       buf,
+                   size_t       len)
+{
+        struct udp_priv * priv = (struct udp_priv *) poa->priv;
+        socklen_t         optlen;
+        size_t            sndbuf = 0;
+        size_t            rcvbuf = 0;
+        int               val;
+        int               size;
+
+        optlen = sizeof(val);
+        if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0)
+                sndbuf = (size_t) val;
+
+        optlen = sizeof(val);
+        if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0)
+                rcvbuf = (size_t) val;
+
+        size = snprintf(buf, len,
+                        "Socket sndbuf (bytes):     %zu\n"
+                        "Socket rcvbuf (bytes):     %zu\n",
+                        sndbuf, rcvbuf);
+        if (size < 0 || (size_t) size >= len)
+                return -1;
+
+        return size;
+}
+
+/*
+ * Asks the kernel which address it would send from: connect() does the
+ * real route lookup and sends nothing, so this honours the default
+ * route, metrics and policy rules alike.
+ */
+static int udp_src_addr(const struct poa_addr * dst,
+                        struct poa_addr *       src)
+{
+        union udp_saddr saddr;
+        socklen_t       len;
+        int             af;
+        int             fd;
+
+        if (dst->type != POA_UDP4 && dst->type != POA_UDP6)
+                return -EINVAL;
+
+        af = dst->type == POA_UDP4 ? AF_INET : AF_INET6;
+
+        addr_to_saddr(dst, &saddr);
+
+        fd = socket(af, SOCK_DGRAM, IPPROTO_UDP);
+        if (fd < 0)
+                return -EIO;
+
+        if (connect(fd, &saddr.sa, saddr_len(af)) < 0)
+                goto fail;
+
+        len = saddr_len(af);
+        if (getsockname(fd, &saddr.sa, &len) < 0)
+                goto fail;
+
+        close(fd);
+
+        saddr_to_addr(&saddr, src);
+
+        return 0;
+
+ fail:
+        close(fd);
+
+        return -EIO;
+}
+
+static void udp_spec(const struct poa * poa,
+                     struct poa_spec *  spec)
+{
+        spec->type = poa->type;
+
+        if (poa->type == POA_UDP4)
+                spec->udp4 = poa->local.udp4;
+        else
+                spec->udp6 = poa->local.udp6;
+}
+
+static bool udp_has_id(const struct poa *      poa,
+                       const struct poa_spec * spec)
+{
+        if (poa->type == POA_UDP4) {
+                if (poa->local.udp4.port != spec->udp4.port)
+                        return false;
+
+                return memcmp(&poa->local.udp4.ip_addr,
+                              &spec->udp4.ip_addr,
+                              sizeof(spec->udp4.ip_addr)) == 0;
+        }
+
+        if (poa->local.udp6.port != spec->udp6.port)
+                return false;
+
+        return memcmp(&poa->local.udp6.ip_addr, &spec->udp6.ip_addr,
+                      sizeof(spec->udp6.ip_addr)) == 0;
+}
+
+static bool udp_addr_is_any(const struct poa_addr * addr)
+{
+        static const struct in6_addr any6 = IN6ADDR_ANY_INIT;
+
+        if (addr->type == POA_UDP4)
+                return addr->udp4.ip_addr.s_addr == htonl(INADDR_ANY);
+
+        return memcmp(&addr->udp6.ip_addr, &any6, sizeof(any6)) == 0;
+}
+
+/*
+ * Our end of the flow: the IP the kernel would send to dst from. A
+ * PoA bound to the wildcard is the catch-all and a failed probe
+ * matches any. Ports are not compared: the probe's is ephemeral.
+ */
+static bool udp_match(const struct poa *      poa,
+                      const struct poa_addr * dst)
+{
+        struct poa_addr src;
+
+        if (udp_addr_is_any(&poa->local))
+                return true;
+
+        if (udp_src_addr(dst, &src) < 0)
+                return true;
+
+        if (poa->type == POA_UDP4)
+                return memcmp(&poa->local.udp4.ip_addr,
+                              &src.udp4.ip_addr,
+                              sizeof(src.udp4.ip_addr)) == 0;
+
+        return memcmp(&poa->local.udp6.ip_addr, &src.udp6.ip_addr,
+                      sizeof(src.udp6.ip_addr)) == 0;
+}
+
+const struct poa_ops udp_poa_ops = {
+        .poa_attach    = udp_attach,
+        .poa_detach    = udp_detach,
+        .poa_start     = udp_start,
+        .poa_stop      = udp_stop,
+        .poa_send      = udp_send,
+        .poa_send_mgmt = udp_send_mgmt,
+        .poa_mtu       = udp_mtu,
+        .poa_qlen      = udp_qlen,
+        .poa_rib       = udp_rib,
+        .poa_spec      = udp_spec,
+        .poa_has_id    = udp_has_id,
+        .poa_match     = udp_match,
+        .mpl           = POA_UDP_MPL
+};
diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c
index 6beae000..aa247755 100644
--- a/src/lib/protobuf.c
+++ b/src/lib/protobuf.c
@@ -22,6 +22,7 @@
 
 #define _DEFAULT_SOURCE
 
+#include <ouroboros/errno.h>
 #include <ouroboros/protobuf.h>
 #include <ouroboros/crypt.h>
 #include <ouroboros/proc.h>
@@ -215,6 +216,166 @@ struct layer_info layer_info_msg_to_s(const layer_info_msg_t * msg)
         return s;
 }
 
+
+static bool mac_is_zero(const uint8_t * mac)
+{
+        static const uint8_t zero[POA_MAC_SIZE] = { 0 };
+
+        return memcmp(mac, zero, POA_MAC_SIZE) == 0;
+}
+
+poa_addr_msg_t * poa_addr_s_to_msg(const struct poa_addr * s)
+{
+        poa_addr_msg_t * msg;
+
+        assert(s != NULL);
+
+        msg = malloc(sizeof(*msg));
+        if (msg == NULL)
+                goto fail_malloc;
+
+        poa_addr_msg__init(msg);
+
+        msg->type = s->type;
+
+        if (s->hostname[0] != '\0') {
+                msg->hostname = strdup(s->hostname);
+                if (msg->hostname == NULL)
+                        goto fail_msg;
+        }
+
+        switch (s->type) {
+        case POA_UDP4:
+                msg->has_ip4  = true;
+                msg->ip4      = s->udp4.ip_addr.s_addr;
+                msg->has_port = true;
+                msg->port     = s->udp4.port;
+                break;
+        case POA_UDP6:
+                msg->has_ip6  = true;
+                msg->ip6.len  = sizeof(s->udp6.ip_addr);
+                msg->ip6.data = malloc(msg->ip6.len);
+                if (msg->ip6.data == NULL)
+                        goto fail_msg;
+                memcpy(msg->ip6.data, &s->udp6.ip_addr, msg->ip6.len);
+
+                msg->has_port = true;
+                msg->port     = s->udp6.port;
+                break;
+        case POA_ETH:
+                msg->has_ethertype = true;
+                msg->ethertype     = s->eth.dst.ethertype;
+                msg->has_mac  = true;
+                msg->mac.len  = POA_MAC_SIZE;
+                msg->mac.data = malloc(POA_MAC_SIZE);
+                if (msg->mac.data == NULL)
+                        goto fail_msg;
+                memcpy(msg->mac.data, s->eth.dst.mac, POA_MAC_SIZE);
+
+                if (!mac_is_zero(s->eth.src.mac)) {
+                        msg->has_src_mac  = true;
+                        msg->src_mac.len  = POA_MAC_SIZE;
+                        msg->src_mac.data = malloc(POA_MAC_SIZE);
+                        if (msg->src_mac.data == NULL)
+                                goto fail_msg;
+                        memcpy(msg->src_mac.data, s->eth.src.mac, POA_MAC_SIZE);
+                }
+
+                if (s->eth.src.dev[0] != '\0') {
+                        msg->dev = strdup(s->eth.src.dev);
+                        if (msg->dev == NULL)
+                                goto fail_msg;
+                }
+                break;
+        case POA_UDP:
+                msg->has_port = true;
+                msg->port     = s->udp4.port;
+                break;
+        default:
+                goto fail_msg;
+        }
+
+        return msg;
+
+ fail_msg:
+        poa_addr_msg__free_unpacked(msg, NULL);
+ fail_malloc:
+        return NULL;
+}
+
+struct poa_addr poa_addr_msg_to_s(const poa_addr_msg_t * msg)
+{
+        struct poa_addr s;
+
+        assert(msg != NULL);
+
+        memset(&s, 0, sizeof(s));
+
+        s.type = POA_INVALID;
+
+        if (msg->hostname != NULL) {
+                if (strlen(msg->hostname) > POA_HOST_STRLEN)
+                        return s;
+                strcpy(s.hostname, msg->hostname);
+        }
+
+        switch (msg->type) {
+        case POA_UDP4:
+                if (!msg->has_ip4 || !msg->has_port)
+                        return s;
+
+                s.udp4.ip_addr.s_addr = msg->ip4;
+                s.udp4.port      = msg->port;
+                break;
+        case POA_UDP6:
+                if (!msg->has_ip6 || !msg->has_port)
+                        return s;
+
+                if (msg->ip6.len != sizeof(s.udp6.ip_addr)
+                    || msg->ip6.data == NULL)
+                        return s;
+
+                memcpy(&s.udp6.ip_addr, msg->ip6.data, msg->ip6.len);
+
+                s.udp6.port = msg->port;
+                break;
+        case POA_ETH:
+                if (!msg->has_ethertype)
+                        return s;
+
+                s.eth.src.ethertype = msg->ethertype;
+                s.eth.dst.ethertype = msg->ethertype;
+
+                if (msg->mac.len != POA_MAC_SIZE || msg->mac.data == NULL)
+                        return s;
+
+                memcpy(s.eth.dst.mac, msg->mac.data, POA_MAC_SIZE);
+
+                if (msg->src_mac.len == POA_MAC_SIZE)
+                        memcpy(s.eth.src.mac, msg->src_mac.data, POA_MAC_SIZE);
+
+                if (msg->dev != NULL) {
+                        if (strlen(msg->dev) > DEV_NAME_SIZE)
+                                return s;
+                        strcpy(s.eth.src.dev, msg->dev);
+                }
+                break;
+        case POA_UDP:
+                if (msg->hostname == NULL || msg->hostname[0] == '\0'
+                    || !msg->has_port)
+                        return s;
+
+                s.udp4.port = msg->port;
+                break;
+        default:
+                return s;
+        }
+
+        s.type = msg->type;
+
+        return s;
+}
+
 ipcp_info_msg_t * ipcp_info_s_to_msg(const struct ipcp_info * s)
 {
         ipcp_info_msg_t * msg;
@@ -389,22 +550,6 @@ struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg)
         return s;
 }
 
-struct dir_dht_config dir_dht_config_msg_to_s(const dir_dht_config_msg_t * msg)
-{
-        struct dir_dht_config s;
-
-        assert(msg != NULL);
-
-        s.params.alpha       = msg->alpha;
-        s.params.k           = msg->k;
-        s.params.t_expire    = msg->t_expire;
-        s.params.t_refresh   = msg->t_refresh;
-        s.params.t_replicate = msg->t_replicate;
-        s.peer               = msg->peer;
-
-        return s;
-}
-
 dir_dht_config_msg_t * dir_dht_config_s_to_msg(const struct dir_dht_config * s)
 {
         dir_dht_config_msg_t * msg;
@@ -427,22 +572,18 @@ dir_dht_config_msg_t * dir_dht_config_s_to_msg(const struct dir_dht_config * s)
         return msg;
 }
 
-struct dir_config dir_config_msg_to_s(const dir_config_msg_t * msg)
+struct dir_dht_config dir_dht_config_msg_to_s(const dir_dht_config_msg_t * msg)
 {
-        struct dir_config s;
+        struct dir_dht_config s;
 
         assert(msg != NULL);
 
-        switch (msg->pol) {
-        case DIR_DHT:
-                s.dht = dir_dht_config_msg_to_s(msg->dht);
-                break;
-        default:
-                /* No checks here */
-                break;
-        }
-
-        s.pol = msg->pol;
+        s.params.alpha       = msg->alpha;
+        s.params.k           = msg->k;
+        s.params.t_expire    = msg->t_expire;
+        s.params.t_refresh   = msg->t_refresh;
+        s.params.t_replicate = msg->t_replicate;
+        s.peer               = msg->peer;
 
         return s;
 }
@@ -479,6 +620,25 @@ dir_config_msg_t * dir_config_s_to_msg(const struct dir_config * s)
         return NULL;
 }
 
+struct dir_config dir_config_msg_to_s(const dir_config_msg_t * msg)
+{
+        struct dir_config s;
+
+        assert(msg != NULL);
+
+        switch (msg->pol) {
+        case DIR_DHT:
+                s.dht = dir_dht_config_msg_to_s(msg->dht);
+                break;
+        default:
+                /* No checks here */
+                break;
+        }
+
+        s.pol = msg->pol;
+
+        return s;
+}
 
 uni_config_msg_t * uni_config_s_to_msg(const struct uni_config * s)
 {
@@ -525,9 +685,9 @@ struct uni_config uni_config_msg_to_s(const uni_config_msg_t * msg)
         return s;
 }
 
-udp4_config_msg_t * udp4_config_s_to_msg(const struct udp4_config * s)
+udp4_poa_msg_t * udp4_poa_s_to_msg(const struct udp4_poa * s)
 {
-        udp4_config_msg_t * msg;
+        udp4_poa_msg_t * msg;
 
         assert(s != NULL);
 
@@ -535,32 +695,30 @@ udp4_config_msg_t * udp4_config_s_to_msg(const struct udp4_config * s)
         if (msg == NULL)
                 return NULL;
 
-        udp4_config_msg__init(msg);
+        udp4_poa_msg__init(msg);
 
-        msg->ip_addr  = s->ip_addr.s_addr;
-        msg->dns_addr = s->dns_addr.s_addr;
-        msg->port     = s->port;
+        msg->ip_addr = s->ip_addr.s_addr;
+        msg->port    = s->port;
 
         return msg;
 }
 
-struct udp4_config udp4_config_msg_to_s(const udp4_config_msg_t * msg)
+struct udp4_poa udp4_poa_msg_to_s(const udp4_poa_msg_t * msg)
 {
-        struct udp4_config s;
+        struct udp4_poa s;
 
         assert(msg != NULL);
 
-        s.ip_addr.s_addr  = msg->ip_addr;
-        s.dns_addr.s_addr = msg->dns_addr;
-        s.port            = msg->port;
+        s.ip_addr.s_addr = msg->ip_addr;
+        s.port           = msg->port;
 
         return s;
 }
 
 #define IN6_LEN (size_t) sizeof(struct in6_addr)
-udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s)
+udp6_poa_msg_t * udp6_poa_s_to_msg(const struct udp6_poa * s)
 {
-        udp6_config_msg_t * msg;
+        udp6_poa_msg_t * msg;
 
         assert(s != NULL);
 
@@ -568,7 +726,7 @@ udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s)
         if (msg == NULL)
                 goto fail_malloc;
 
-        udp6_config_msg__init(msg);
+        udp6_poa_msg__init(msg);
 
         msg->ip_addr.data = malloc(IN6_LEN);
         if (msg->ip_addr.data == NULL)
@@ -577,42 +735,33 @@ udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s)
         msg->ip_addr.len = IN6_LEN;
         memcpy(msg->ip_addr.data, &s->ip_addr.s6_addr, IN6_LEN);
 
-        msg->dns_addr.data = malloc(IN6_LEN);
-        if (msg->dns_addr.data == NULL)
-                goto fail_msg;
-
-        msg->dns_addr.len = IN6_LEN;
-        memcpy(msg->dns_addr.data, &s->dns_addr.s6_addr, IN6_LEN);
-
         msg->port = s->port;
 
         return msg;
 
  fail_msg:
-        udp6_config_msg__free_unpacked(msg, NULL);
+        udp6_poa_msg__free_unpacked(msg, NULL);
  fail_malloc:
         return NULL;
 }
 
-struct udp6_config udp6_config_msg_to_s(const udp6_config_msg_t * msg)
+struct udp6_poa udp6_poa_msg_to_s(const udp6_poa_msg_t * msg)
 {
-        struct udp6_config s;
+        struct udp6_poa s;
 
         assert(msg != NULL);
 
         assert(msg->ip_addr.len == IN6_LEN);
-        assert(msg->dns_addr.len == IN6_LEN);
 
         memcpy(&s.ip_addr.s6_addr, msg->ip_addr.data, IN6_LEN);
-        memcpy(&s.dns_addr.s6_addr, msg->dns_addr.data, IN6_LEN);
         s.port = msg->port;
 
         return s;
 }
 
-eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s)
+eth_poa_msg_t * eth_poa_s_to_msg(const struct eth_poa * s)
 {
-        eth_config_msg_t * msg;
+        eth_poa_msg_t * msg;
 
         assert(s != NULL);
 
@@ -620,7 +769,7 @@ eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s)
         if (msg == NULL)
                 goto fail_malloc;
 
-        eth_config_msg__init(msg);
+        eth_poa_msg__init(msg);
 
         msg->dev = strdup(s->dev);
         if (msg->dev == NULL)
@@ -628,28 +777,115 @@ eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s)
 
         msg->ethertype = s->ethertype;
 
+        msg->has_mac = true;
+        msg->mac.len = POA_MAC_SIZE;
+        msg->mac.data = malloc(POA_MAC_SIZE);
+        if (msg->mac.data == NULL)
+                goto fail_msg;
+
+        memcpy(msg->mac.data, s->mac, POA_MAC_SIZE);
+
         return msg;
 
  fail_msg:
-        eth_config_msg__free_unpacked(msg, NULL);
+        eth_poa_msg__free_unpacked(msg, NULL);
  fail_malloc:
         return NULL;
 }
 
-struct eth_config eth_config_msg_to_s(const eth_config_msg_t * msg)
+struct eth_poa eth_poa_msg_to_s(const eth_poa_msg_t * msg)
 {
-        struct eth_config s;
+        struct eth_poa s;
 
         assert(msg != NULL);
         assert(strlen(msg->dev) <= DEV_NAME_SIZE);
 
+        memset(&s, 0, sizeof(s));
+
         strcpy(s.dev, msg->dev);
+
         s.ethertype = msg->ethertype;
 
+        if (msg->mac.len == POA_MAC_SIZE)
+                memcpy(s.mac, msg->mac.data, POA_MAC_SIZE);
+
         return s;
 }
 
 
+poa_spec_msg_t * poa_spec_s_to_msg(const struct poa_spec * s)
+{
+        poa_spec_msg_t * msg;
+
+        assert(s != NULL);
+
+        msg = malloc(sizeof(*msg));
+        if (msg == NULL)
+                goto fail_malloc;
+
+        poa_spec_msg__init(msg);
+
+        switch (s->type) {
+        case POA_UDP4:
+                msg->udp4 = udp4_poa_s_to_msg(&s->udp4);
+                if (msg->udp4 == NULL)
+                        goto fail_msg;
+                break;
+        case POA_UDP6:
+                msg->udp6 = udp6_poa_s_to_msg(&s->udp6);
+                if (msg->udp6 == NULL)
+                        goto fail_msg;
+                break;
+        case POA_ETH:
+                msg->eth = eth_poa_s_to_msg(&s->eth);
+                if (msg->eth == NULL)
+                        goto fail_msg;
+                break;
+        default:
+                goto fail_msg;
+        }
+
+        return msg;
+
+ fail_msg:
+        poa_spec_msg__free_unpacked(msg, NULL);
+ fail_malloc:
+        return NULL;
+}
+
+struct poa_spec poa_spec_msg_to_s(const poa_spec_msg_t * msg)
+{
+        struct poa_spec s;
+
+        memset(&s, 0, sizeof(s));
+
+        s.type = POA_INVALID;
+
+        if (msg == NULL)
+                return s;
+
+        if (msg->udp4 != NULL) {
+                s.type = POA_UDP4;
+                s.udp4 = udp4_poa_msg_to_s(msg->udp4);
+        } else if (msg->udp6 != NULL) {
+                if (msg->udp6->ip_addr.len != IN6_LEN
+                    || msg->udp6->ip_addr.data == NULL)
+                        return s;
+
+                s.type = POA_UDP6;
+                s.udp6 = udp6_poa_msg_to_s(msg->udp6);
+        } else if (msg->eth != NULL) {
+                if (msg->eth->dev == NULL
+                    || strlen(msg->eth->dev) > DEV_NAME_SIZE)
+                        return s;
+
+                s.type = POA_ETH;
+                s.eth  = eth_poa_msg_to_s(msg->eth);
+        }
+
+        return s;
+}
+
 ipcp_config_msg_t * ipcp_config_s_to_msg(const struct ipcp_config * s)
 {
         ipcp_config_msg_t * msg;
@@ -672,23 +908,6 @@ ipcp_config_msg_t * ipcp_config_s_to_msg(const struct ipcp_config * s)
                 break;
         case IPCP_BROADCAST:
                 break;
-        case IPCP_ETH_LLC:
-                /* FALLTHRU */
-        case IPCP_ETH_DIX:
-                msg->eth = eth_config_s_to_msg(&s->eth);
-                if (msg->eth == NULL)
-                        goto fail_msg;
-                break;
-        case IPCP_UDP4:
-                msg->udp4 = udp4_config_s_to_msg(&s->udp4);
-                if (msg->udp4 == NULL)
-                        goto fail_msg;
-                break;
-        case IPCP_UDP6:
-                msg->udp6 = udp6_config_s_to_msg(&s->udp6);
-                if (msg->udp6 == NULL)
-                        goto fail_msg;
-                break;
         default:
                 /* No checks here */
                 break;
@@ -714,6 +933,8 @@ struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg)
 
         assert(msg != NULL);
 
+        memset(&s, 0, sizeof(s));
+
         s.type = msg->ipcp_type;
 
         s.layer_info = layer_info_msg_to_s(msg->layer_info);
@@ -724,17 +945,6 @@ struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg)
         case IPCP_UNICAST:
                 s.unicast = uni_config_msg_to_s(msg->unicast);
                 break;
-        case IPCP_ETH_LLC:
-                /* FALLTHRU */
-        case IPCP_ETH_DIX:
-                s.eth = eth_config_msg_to_s(msg->eth);
-                break;
-        case IPCP_UDP4:
-                s.udp4 = udp4_config_msg_to_s(msg->udp4);
-                break;
-        case IPCP_UDP6:
-                s.udp6 = udp6_config_msg_to_s(msg->udp6);
-                break;
         case IPCP_BROADCAST:
                 break;
         default:
diff --git a/src/lib/serdes-irm.c b/src/lib/serdes-irm.c
index 1d9b4dec..20b25a1b 100644
--- a/src/lib/serdes-irm.c
+++ b/src/lib/serdes-irm.c
@@ -441,6 +441,7 @@ int proc_exit__irm_req_ser(buffer_t *   buf)
         return -ENOMEM;
 }
 
+/* data is borrowed from the caller; detach before free. */
 int ipcp_flow_req_arr__irm_req_ser(buffer_t *               buf,
                                    const buffer_t *         dst,
                                    const struct flow_info * flow,
@@ -494,6 +495,138 @@ int ipcp_flow_req_arr__irm_req_ser(buffer_t *               buf,
         return -ENOMEM;
 }
 
+static int __ep_flow_ser(buffer_t *               buf,
+                         const struct flow_info * flow,
+                         const buffer_t *         data,
+                         int                      response,
+                         const char *             dst,
+                         int                      code)
+{
+        irm_msg_t * msg;
+        size_t      len;
+        int         err = -ENOMEM;
+
+        msg = malloc(sizeof(*msg));
+        if (msg == NULL)
+                goto fail_malloc;
+
+        irm_msg__init(msg);
+
+        msg->code      = code;
+        msg->flow_info = flow_info_s_to_msg(flow);
+        if (msg->flow_info == NULL)
+                goto fail_msg;
+
+        if (dst != NULL) {
+                msg->dst = strdup(dst);
+                if (msg->dst == NULL)
+                        goto fail_msg;
+        }
+
+        if (data != NULL) {
+                msg->has_pk  = true;
+                msg->pk.len  = data->len;
+                msg->pk.data = data->data;
+        }
+
+        msg->has_response = true;
+        msg->response     = response;
+
+        len = irm_msg__get_packed_size(msg);
+        if (len == 0 || len > buf->len)
+                goto fail_msg;
+
+        buf->len = len;
+
+        irm_msg__pack(msg, buf->data);
+
+        err = 0;
+ fail_msg:
+        msg->pk.len  = 0;
+        msg->pk.data = NULL;
+
+        irm_msg__free_unpacked(msg, NULL);
+ fail_malloc:
+        return err;
+}
+
+int poa_flow_alloc__irm_req_ser(buffer_t *               buf,
+                                const struct flow_info * flow,
+                                const char *             dst)
+{
+        return __ep_flow_ser(buf, flow, NULL, 0, dst,
+                             IRM_MSG_CODE__IRM_POA_FLOW_ALLOC);
+}
+
+int poa_flow_alloc_r__irm_req_ser(buffer_t *               buf,
+                                  const struct flow_info * flow,
+                                  const buffer_t *         data,
+                                  int                      response)
+{
+        return __ep_flow_ser(buf, flow, data, response, NULL,
+                             IRM_MSG_CODE__IRM_POA_FLOW_ALLOC_R);
+}
+
+int ipcp_poa_flow_req_arr__irm_req_ser(buffer_t *               buf,
+                                       const struct flow_info * flow,
+                                       const buffer_t *         data)
+{
+        return __ep_flow_ser(buf, flow, data, 0, NULL,
+                             IRM_MSG_CODE__IPCP_POA_FLOW_REQ_ARR);
+}
+
+int poa_flow__irm_result_des(buffer_t *         buf,
+                             struct flow_info * flow,
+                             buffer_t *         data)
+{
+        irm_msg_t * msg;
+        int         err;
+
+        msg = irm_msg__unpack(NULL, buf->len, buf->data);
+        if (msg == NULL) {
+                err = -EIRMD;
+                goto fail_msg;
+        }
+
+        if (!msg->has_result) {
+                err = -EIRMD;
+                goto fail;
+        }
+
+        if (msg->result < 0) {
+                err = msg->result;
+                goto fail;
+        }
+
+        if (msg->flow_info == NULL) {
+                err = -EBADF;
+                goto fail;
+        }
+
+        *flow = flow_info_msg_to_s(msg->flow_info);
+
+        clrbuf(*data);
+
+        if (msg->has_pk && msg->pk.len > 0) {
+                data->data = malloc(msg->pk.len);
+                if (data->data == NULL) {
+                        err = -ENOMEM;
+                        goto fail;
+                }
+                memcpy(data->data, msg->pk.data, msg->pk.len);
+
+                data->len = msg->pk.len;
+        }
+
+        irm_msg__free_unpacked(msg, NULL);
+
+        return 0;
+ fail:
+        irm_msg__free_unpacked(msg, NULL);
+ fail_msg:
+        return err;
+}
+
 int ipcp_flow_update_arr__irm_req_ser(buffer_t *               buf,
                                       const struct flow_info * flow,
                                       const buffer_t *         data)
diff --git a/src/lib/tests/CMakeLists.txt b/src/lib/tests/CMakeLists.txt
index 002d94af..1f2e9ba2 100644
--- a/src/lib/tests/CMakeLists.txt
+++ b/src/lib/tests/CMakeLists.txt
@@ -11,6 +11,7 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c
   bitmap_test.c
   btree_test.c
   crypt_test.c
+  poa_test.c
   hash_test.c
   kex_test.c
   kex_test_ml_kem.c
@@ -25,6 +26,12 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c
 
 add_executable(${PARENT_DIR}_test ${${PARENT_DIR}_tests})
 
+if(HAVE_LIBURCU)
+  # poa_test.c pulls in poa.h, whose urcu guard needs C99.
+  set_source_files_properties(poa_test.c PROPERTIES
+    COMPILE_OPTIONS "-std=gnu99")
+endif()
+
 target_include_directories(${PARENT_DIR}_test PRIVATE
   ${CMAKE_SOURCE_DIR}/src/lib)
 
diff --git a/src/lib/tests/poa_test.c b/src/lib/tests/poa_test.c
new file mode 100644
index 00000000..99886769
--- /dev/null
+++ b/src/lib/tests/poa_test.c
@@ -0,0 +1,307 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2026
+ *
+ * Flow PoA tests
+ *
+ *    Dimitri Staessens <dimitri@ouroboros.rocks>
+ *    Sander Vrijders   <sander@ouroboros.rocks>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+#if defined(__linux__) || defined(__CYGWIN__)
+#define _DEFAULT_SOURCE
+#else
+#define _POSIX_C_SOURCE 200809L
+#endif
+
+#include <test/test.h>
+
+#include "poa/addr.c"
+#ifdef HAVE_RAW_SOCKETS
+#include "poa/eth.c"
+#endif
+
+#include <arpa/inet.h>
+#include <stdio.h>
+
+#ifdef HAVE_RAW_SOCKETS
+static const uint8_t eth_our_mac[POA_MAC_SIZE] =
+        { 0x02, 0x00, 0x00, 0x00, 0x00, 0x01 };
+static const uint8_t eth_far_mac[POA_MAC_SIZE] =
+        { 0x02, 0x00, 0x00, 0x00, 0x00, 0x02 };
+static const uint8_t eth_bc_mac[POA_MAC_SIZE] =
+        { 0xff, 0xff, 0xff, 0xff, 0xff, 0xff };
+#endif
+
+/* PoA core stubs: the reader threads never run in this test. */
+void poa_rx_pkt(struct poa *         poa,
+                uint32_t             eid,
+                struct ssm_pk_buff * spb)
+{
+        (void) poa;
+        (void) eid;
+        (void) spb;
+}
+
+void poa_rx_mgmt(struct poa *            poa,
+                 const struct poa_addr * src,
+                 const uint8_t *         buf,
+                 size_t                  len)
+{
+        (void) poa;
+        (void) src;
+        (void) buf;
+        (void) len;
+}
+
+int poa_spb_reserve(struct ssm_pk_buff ** spb,
+                    size_t                len)
+{
+        (void) spb;
+        (void) len;
+
+        return -1;
+}
+
+size_t poa_link_updown(int  ifindex,
+                       bool up)
+{
+        (void) ifindex;
+        (void) up;
+
+        return 0;
+}
+
+bool poa_has_name(const uint8_t * hash)
+{
+        (void) hash;
+
+        return false;
+}
+
+int poa_bcast_mgmt(const struct poa_addr * dst,
+                   const uint8_t *         buf,
+                   size_t                  len)
+{
+        (void) dst;
+        (void) buf;
+        (void) len;
+
+        return 0;
+}
+
+static void udp4_addr(struct poa_addr * addr,
+                      const char *      ip,
+                      uint16_t          port)
+{
+        memset(addr, 0, sizeof(*addr));
+
+        addr->type      = POA_UDP4;
+        addr->udp4.port = port;
+
+        inet_pton(AF_INET, ip, &addr->udp4.ip_addr);
+}
+
+static int test_poa_addr_cmp(void)
+{
+        struct poa_addr a;
+        struct poa_addr b;
+
+        TEST_START();
+
+        udp4_addr(&a, "10.0.0.10", 3435);
+        udp4_addr(&b, "10.0.0.10", 3435);
+
+        if (poa_addr_cmp(&a, &b) != 0) {
+                printf("Identical addresses did not match.\n");
+                goto fail;
+        }
+
+        udp4_addr(&b, "10.0.0.10", 3436);
+
+        if (poa_addr_cmp(&a, &b) == 0) {
+                printf("Addresses with a different port matched.\n");
+                goto fail;
+        }
+
+        udp4_addr(&b, "10.0.0.11", 3435);
+
+        if (poa_addr_cmp(&a, &b) == 0) {
+                printf("Addresses with a different ip matched.\n");
+                goto fail;
+        }
+
+        b.type = POA_UDP6;
+        if (poa_addr_cmp(&a, &b) == 0) {
+                printf("Addresses of a different type matched.\n");
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+static int test_poa_mgmt_msg_qos(void)
+{
+        struct poa_mgmt_msg msg;
+        qosspec_t           qs;
+
+        TEST_START();
+
+        poa_mgmt_msg_ser(&msg, POA_FLOW_REQ, 1, 2, qos_stream, 0, 0);
+
+        if (msg.code != POA_FLOW_REQ) {
+                printf("Wrong code in management message.\n");
+                goto fail;
+        }
+
+        if (ntoh32(msg.s_eid) != 1 || ntoh32(msg.d_eid) != 2) {
+                printf("Wrong PoA ids in management message.\n");
+                goto fail;
+        }
+
+        memset(&qs, 0, sizeof(qs));
+
+        poa_mgmt_msg_qos(&msg, &qs);
+
+        if (memcmp(&qs, &qos_stream, sizeof(qs)) != 0) {
+                printf("QoS did not survive the management message.\n");
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+#ifdef HAVE_RAW_SOCKETS
+
+static void eth_fake_priv(struct eth_priv * priv,
+                          struct poa *      e)
+{
+        memset(priv, 0, sizeof(*priv));
+        memset(e, 0, sizeof(*e));
+
+        e->type = POA_ETH;
+
+        strcpy(e->local.eth.src.dev, "test0");
+
+        priv->poa       = e;
+        priv->ethertype = htons(0xA000);
+        priv->mtu       = 1500;
+
+        memcpy(priv->hw_addr, eth_our_mac, POA_MAC_SIZE);
+}
+
+static int test_eth_frame(void)
+{
+        struct eth_priv  priv;
+        struct poa       e;
+        uint8_t          buf[64];
+        struct eth_hdr * hdr = (struct eth_hdr *) buf;
+        const char *     body = "hello";
+        size_t           n;
+        size_t           plen;
+        uint32_t         eid;
+
+        TEST_START();
+
+        eth_fake_priv(&priv, &e);
+
+        eth_hdr_ser(&priv, hdr, eth_our_mac, 7, strlen(body));
+
+        memcpy(buf + ETH_HDR_TOT_SIZE, body, strlen(body));
+
+        n = ETH_HDR_TOT_SIZE + strlen(body);
+        if (frame_parse(&priv, buf, n, &eid, &plen) < 0) {
+                printf("Failed to parse a valid frame.\n");
+                goto fail;
+        }
+
+        if (eid != 7 || plen != strlen(body)) {
+                printf("Expected eid 7 len %zu, got %u len %zu.\n",
+                       strlen(body), eid, plen);
+                goto fail;
+        }
+
+        hdr->poa.hcs ^= 0xFF;
+
+        if (frame_parse(&priv, buf, n, &eid, &plen) == 0) {
+                printf("Accepted a corrupt header checksum.\n");
+                goto fail;
+        }
+
+        hdr->poa.hcs ^= 0xFF;
+        hdr->ethertype ^= 0xFF;
+
+        if (frame_parse(&priv, buf, n, &eid, &plen) == 0) {
+                printf("Accepted a foreign Ethertype.\n");
+                goto fail;
+        }
+
+        hdr->ethertype ^= 0xFF;
+
+        if (frame_parse(&priv, buf, ETH_HDR_SIZE + 2, &eid, &plen) == 0) {
+                printf("Accepted a truncated frame.\n");
+                goto fail;
+        }
+
+        eth_hdr_ser(&priv, hdr, eth_far_mac, 7, strlen(body));
+
+        if (frame_parse(&priv, buf, n, &eid, &plen) == 0) {
+                printf("Accepted a frame for another host.\n");
+                goto fail;
+        }
+
+        eth_hdr_ser(&priv, hdr, eth_bc_mac, 7, strlen(body));
+
+        if (frame_parse(&priv, buf, n, &eid, &plen) < 0) {
+                printf("Rejected a broadcast frame.\n");
+                goto fail;
+        }
+
+        TEST_SUCCESS();
+
+        return TEST_RC_SUCCESS;
+ fail:
+        TEST_FAIL();
+        return TEST_RC_FAIL;
+}
+
+#endif /* HAVE_RAW_SOCKETS */
+
+int poa_test(int     argc,
+             char ** argv)
+{
+        int ret = 0;
+
+        (void) argc;
+        (void) argv;
+
+        ret |= test_poa_addr_cmp();
+        ret |= test_poa_mgmt_msg_qos();
+#ifdef HAVE_RAW_SOCKETS
+        ret |= test_eth_frame();
+#endif
+
+        return ret;
+}