Repository: nanomsg/nng (c)

Issue:
Extract and deprecate polyamorous mode
Pair v1 Polyamorous mode was a design mistake.   It adds complexity to the pair protocol, while bringing very little value.  It can't operate over proxies, and it hurts performance.

What is really needed is a mesh protocol which should support broadcast, unicast, and route discovery for directed sends.

It is possible to mostly replace designs using pair v1 with surveyor pattern, where the monogamous peers send extremely long (infinite?) surveys, and the polyamorous peers reply as respondents.  It might require a little more application complexity to do so, and probably raw mode would help.  The advantage of such approaches are that they can interoperate with all nanomsg implementations, and support operation over proxies.

In the meantime we should move deprecate the mode, and move it to its own implementation so that pair1 optimizations can proceed unimpeded.

The option should be deprecated, and a new nng_open_pair1_poly() function provided for now.

Once we have a better mesh replacement, the pair1 mode can be removed.

Pull request description:
fixes #1189 Extract and deprecate polyamorous mode
See the bug report for details.

To use pair1 polyamorous mode after this, code has to be converted to use nng_pair1_open_poly().

Raw mode polyamorous sockets are not supported.

Full test coverage is included.  As a bonus, we've improved the coverage for pair1 itself.

Patch:
diff --git a/docs/man/nng_pair.7.adoc b/docs/man/nng_pair.7.adoc
index dff13e8e..62075753 100644
--- a/docs/man/nng_pair.7.adoc
+++ b/docs/man/nng_pair.7.adoc
@@ -49,10 +49,6 @@ Applications that require reliable delivery semantics should consider using
 xref:nng_req.7.adoc[_req_] sockets, or
 implement their own acknowledgment layer on top of _pair_ sockets.
 
-In order to avoid head-of-line blocking conditions, _polyamorous_ mode _pair_
-sockets (version 1 only) discard messages if they are unable to deliver them
-to a peer.
-
 === Protocol Versions
 
 Version 0 is the legacy version of this protocol.
@@ -66,34 +62,39 @@ https://github.com/go-mangos/mangos[mangos].
 Version 1 of the protocol offers improved protection against loops when
 used with xref:nng_device.3.adoc[`nng_device()`].
 
-NOTE: Version 1 of this protocol is considered experimental at this time.
-
 === Polyamorous Mode
 
-CAUTION: Polyamorous mode was provided as an experiment, and that experiment
-is deemed to have failed.
-Thus use of this mode is discouraged.
-In particular, the design cannot function over
-xref:nng_device.3.adoc[device] proxies.
-Consequently, this mode may be removed in a future release.
+WARNING: Polyamorous mode is deprecated, and support for it will likely
+be removed in a future release.
+Applications are strongly discouraged from making further use of it.
 
 Normally pair sockets are for one-to-one communication, and a given peer
 will reject new connections if it already has an active connection to another
 peer.
 
-In ((_polyamorous_ mode)), which is only available with version 1, a socket can
-support many one-to-one connections.
+((_Polyamorous_ mode)) changes this, to allow a socket to communicate with
+multiple directly-connected peers.
+This mode is enabled by opening a socket with the
+xref:nng_pair_open.3.adoc[`nng_pair1_open_poly()`]
+function
+
+NOTE: Polyamorous mode is only available when using pair version 1.
+
+In polyamorous mode a socket can support many one-to-one connections.
 In this mode, the application must
 choose the remote peer to receive an outgoing message by setting the
 xref:nng_pipe.5.adoc[`nng_pipe`] to use for the outgoing message with
 the xref:nng_msg_set_pipe.3.adoc[`nng_msg_set_pipe()`] function.
 
+If no remote peer is specified by the sender, then the protocol will select
+any available connected peer.
+
 Most often the value of the outgoing pipe will be obtained from an incoming
 message using the xref:nng_msg_get_pipe.3.adoc[`nng_msg_get_pipe()`] function,
 such as when replying to an incoming message.
 
-NOTE: This capability _only_ works with directly connected peers.
-It is not possible to use polyamorous mode in the presence of proxies.
+NOTE: Directed send _only_ works with directly connected peers.
+It will not function across xref:nng_device.3.adoc[device] proxies.
 
 In order to prevent head-of-line blocking, if the peer on the given pipe
 is not able to receive (or the pipe is no longer available, such as if the
@@ -104,19 +105,16 @@ to the sender.
 
 The following protocol-specific options are available.
 
-((`NNG_OPT_PAIR1_POLY`))::
-
-   (`bool`, version 1 only)  This option enables the use of _polyamorous_ mode.
-   The value is read-write, and takes an integer Boolean value.  The default
-   false value (0) indicates that legacy monogamous mode should be used.
-
-NOTE: This option, and the associated capability are considered failed experiments.
-Thus, use of it is discouraged, and it may be removed in a future release.
-
 xref:nng_options.5.adoc#NNG_OPT_MAXTTL[`NNG_OPT_MAXTTL`]::
 
    (`int`, version 1 only).  Maximum time-to-live.
 
+((`NNG_OPT_PAIR1_POLY`))::
+
+   (`bool`, version 1 only)  This option is no longer supported.
+   Formerly it was used to configure _polyamorous_ mode, but that mode
+   is now established by using the `nng_pair1_open_poly()` function.
+
 === Protocol Headers
 
 Version 0 of the pair protocol has no protocol-specific headers.
diff --git a/docs/man/nng_pair_open.3.adoc b/docs/man/nng_pair_open.3.adoc
index c61e6110..20348123 100644
--- a/docs/man/nng_pair_open.3.adoc
+++ b/docs/man/nng_pair_open.3.adoc
@@ -33,6 +33,8 @@ int nng_pair0_open_raw(nng_socket *s);
 int nng_pair1_open(nng_socket *s);
 
 int nng_pair1_open_raw(nng_socket *s);
+
+int nng_pair1_open_poly(nng_socktet *s);
 ----
 
 == DESCRIPTION
@@ -46,6 +48,12 @@ create a xref:nng_pair.7.adoc[_pair_] version 0 or version 1
 xref:nng_socket.5.adoc[socket] in
 xref:nng.7.adoc#raw_mode[raw] mode and return it at the location pointed to by _s_.
 
+The `nng_pair1_open_poly()` function opens a pair version 1 socket in
+polyamorous mode.
+
+NOTE: Polyamorous mode is deprecated and should not be used in new applications.
+The `nng_pair1_open_poly()` function will likely be removed in a future release.
+
 == RETURN VALUES
 
 These functions returns 0 on success, and non-zero otherwise.
diff --git a/include/nng/protocol/pair1/pair.h b/include/nng/protocol/pair1/pair.h
index b89f3105..c48b2021 100644
--- a/include/nng/protocol/pair1/pair.h
+++ b/include/nng/protocol/pair1/pair.h
@@ -1,5 +1,5 @@
 //
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
 //
 // This software is supplied under the terms of the MIT License, a
@@ -17,6 +17,7 @@ extern "C" {
 
 NNG_DECL int nng_pair1_open(nng_socket *);
 NNG_DECL int nng_pair1_open_raw(nng_socket *);
+NNG_DECL int nng_pair1_open_poly(nng_socket *);
 
 #ifndef nng_pair_open
 #define nng_pair_open nng_pair1_open
diff --git a/src/protocol/pair1/CMakeLists.txt b/src/protocol/pair1/CMakeLists.txt
index 3f92c585..f586fabe 100644
--- a/src/protocol/pair1/CMakeLists.txt
+++ b/src/protocol/pair1/CMakeLists.txt
@@ -12,7 +12,9 @@
 option (NNG_PROTO_PAIR1 "Enable PAIRv1 protocol." ON)
 mark_as_advanced(NNG_PROTO_PAIR1)
 
-nng_sources_if(NNG_PROTO_PAIR1 pair.c)
+# XXX: break pair1_poly into an ifdef.
+nng_sources_if(NNG_PROTO_PAIR1 pair.c pair1_poly.c)
 nng_headers_if(NNG_PROTO_PAIR1 nng/protocol/pair1/pair.h)
 nng_defines_if(NNG_PROTO_PAIR1 NNG_HAVE_PAIR1)
-nng_test(pair1_test)
\ No newline at end of file
+nng_test(pair1_test)
+nng_test(pair1_poly_test)
\ No newline at end of file
diff --git a/src/protocol/pair1/pair.c b/src/protocol/pair1/pair.c
index f564195d..b98975c3 100644
--- a/src/protocol/pair1/pair.c
+++ b/src/protocol/pair1/pair.c
@@ -13,16 +13,13 @@
 #include "core/nng_impl.h"
 #include "nng/protocol/pair1/pair.h"
 
-// Pair protocol.  The PAIRv1 protocol is a simple 1:1 messaging pattern,
-// usually, but it can support a polyamorous mode where a single server can
-// communicate with multiple partners.
+// Pair protocol.  The PAIRv1 protocol is a simple 1:1 messaging pattern.
 
 #define BUMP_STAT(x) nni_stat_inc_atomic(x, 1)
 
 typedef struct pair1_pipe pair1_pipe;
 typedef struct pair1_sock pair1_sock;
 
-static void pair1_sock_get_cb(void *);
 static void pair1_pipe_send_cb(void *);
 static void pair1_pipe_recv_cb(void *);
 static void pair1_pipe_get_cb(void *);
@@ -40,8 +37,6 @@ struct pair1_sock {
 	nni_idhash *   pipes;
 	nni_list       plist;
 	bool           started;
-	bool           poly;
-	nni_aio        aio_get;
 	nni_stat_item  stat_poly;
 	nni_stat_item  stat_raw;
 	nni_stat_item  stat_reject_mismatch;
@@ -50,13 +45,15 @@ struct pair1_sock {
 	nni_stat_item  stat_rx_malformed;
 	nni_stat_item  stat_tx_malformed;
 	nni_stat_item  stat_tx_drop;
+#ifdef NNG_TEST_LIB
+	bool inject_header;
+#endif
 };
 
 // pair1_pipe is our per-pipe protocol private structure.
 struct pair1_pipe {
 	nni_pipe *    pipe;
 	pair1_sock *  pair;
-	nni_msgq *    send_queue;
 	nni_aio       aio_send;
 	nni_aio       aio_recv;
 	nni_aio       aio_get;
@@ -69,7 +66,6 @@ pair1_sock_fini(void *arg)
 {
 	pair1_sock *s = arg;
 
-	nni_aio_fini(&s->aio_get);
 	nni_idhash_fini(s->pipes);
 	nni_mtx_fini(&s->mtx);
 }
@@ -87,11 +83,8 @@ pair1_sock_init_impl(void *arg, nni_sock *sock, bool raw)
 	// Raw mode uses this.
 	nni_mtx_init(&s->mtx);
 
-	nni_aio_init(&s->aio_get, pair1_sock_get_cb, s);
-
 	nni_stat_init_bool(
 	    &s->stat_poly, "polyamorous", "polyamorous mode?", false);
-	nni_stat_set_lock(&s->stat_poly, &s->mtx);
 	nni_sock_add_stat(sock, &s->stat_poly);
 
 	nni_stat_init_bool(&s->stat_raw, "raw", "raw mode?", raw);
@@ -131,7 +124,6 @@ pair1_sock_init_impl(void *arg, nni_sock *sock, bool raw)
 
 	s->sock = sock;
 	s->raw  = raw;
-	s->poly = false;
 	s->uwq  = nni_sock_sendq(sock);
 	s->urq  = nni_sock_recvq(sock);
 	nni_atomic_init(&s->ttl);
@@ -172,25 +164,18 @@ pair1_pipe_fini(void *arg)
 	nni_aio_fini(&p->aio_recv);
 	nni_aio_fini(&p->aio_put);
 	nni_aio_fini(&p->aio_get);
-	nni_msgq_fini(p->send_queue);
 }
 
 static int
 pair1_pipe_init(void *arg, nni_pipe *pipe, void *pair)
 {
 	pair1_pipe *p = arg;
-	int         rv;
 
 	nni_aio_init(&p->aio_send, pair1_pipe_send_cb, p);
 	nni_aio_init(&p->aio_recv, pair1_pipe_recv_cb, p);
 	nni_aio_init(&p->aio_get, pair1_pipe_get_cb, p);
 	nni_aio_init(&p->aio_put, pair1_pipe_put_cb, p);
 
-	if ((rv = nni_msgq_init(&p->send_queue, 2)) != 0) {
-		pair1_pipe_fini(p);
-		return (rv);
-	}
-
 	p->pipe = pipe;
 	p->pair = pair;
 
@@ -218,31 +203,19 @@ pair1_pipe_start(void *arg)
 		nni_mtx_unlock(&s->mtx);
 		return (rv);
 	}
-	if (!s->poly) {
-		if (!nni_list_empty(&s->plist)) {
-			nni_idhash_remove(s->pipes, id);
-			nni_mtx_unlock(&s->mtx);
-			BUMP_STAT(&s->stat_reject_already);
-			return (NNG_EBUSY);
-		}
-	} else {
-		if (!s->started) {
-			nni_msgq_aio_get(s->uwq, &s->aio_get);
-		}
+	if (!nni_list_empty(&s->plist)) {
+		nni_idhash_remove(s->pipes, id);
+		nni_mtx_unlock(&s->mtx);
+		BUMP_STAT(&s->stat_reject_already);
+		return (NNG_EBUSY);
 	}
 	nni_list_append(&s->plist, p);
 	s->started = true;
 	nni_mtx_unlock(&s->mtx);
 
-	// Schedule a get.  In polyamorous mode we get on the per pipe
-	// send_queue, as the socket distributes to us. In monogamous mode
-	// we bypass and get from the upper write queue directly (saving a
-	// set of context switches).
-	if (s->poly) {
-		nni_msgq_aio_get(p->send_queue, &p->aio_get);
-	} else {
-		nni_msgq_aio_get(s->uwq, &p->aio_get);
-	}
+	// Schedule a get.
+	nni_msgq_aio_get(s->uwq, &p->aio_get);
+
 	// And the pipe read of course.
 	nni_pipe_recv(p->pipe, &p->aio_recv);
 
@@ -264,8 +237,6 @@ pair1_pipe_close(void *arg)
 	nni_idhash_remove(s->pipes, nni_pipe_id(p->pipe));
 	nni_list_node_remove(&p->node);
 	nni_mtx_unlock(&s->mtx);
-
-	nni_msgq_close(p->send_queue);
 }
 
 static void
@@ -318,47 +289,6 @@ pair1_pipe_recv_cb(void *arg)
 	nni_msgq_aio_put(s->urq, &p->aio_put);
 }
 
-static void
-pair1_sock_get_cb(void *arg)
-{
-	pair1_pipe *p;
-	pair1_sock *s = arg;
-	nni_msg *   msg;
-	uint32_t    id;
-
-	if (nni_aio_result(&s->aio_get) != 0) {
-		// Socket closing...
-		return;
-	}
-
-	msg = nni_aio_get_msg(&s->aio_get);
-	nni_aio_set_msg(&s->aio_get, NULL);
-
-	p = NULL;
-	nni_mtx_lock(&s->mtx);
-	// By definition we are in polyamorous mode.
-	NNI_ASSERT(s->poly);
-	// If no pipe was requested, we look for any connected peer.
-	if (((id = nni_msg_get_pipe(msg)) == 0) &&
-	    (!nni_list_empty(&s->plist))) {
-		p = nni_list_first(&s->plist);
-	} else {
-		nni_idhash_find(s->pipes, id, (void **) &p);
-	}
-
-	// Try a non-blocking send.  If this fails we just discard the
-	// message.  We have to do this to avoid head-of-line blocking
-	// for messages sent to other pipes.  Note that there is some
-	// buffering in the send_queue.
-	if ((p == NULL) || nni_msgq_tryput(p->send_queue, msg) != 0) {
-		BUMP_STAT(&s->stat_tx_drop);
-		nni_msg_free(msg);
-	}
-
-	nni_mtx_unlock(&s->mtx);
-	nni_msgq_aio_get(s->uwq, &s->aio_get);
-}
-
 static void
 pair1_pipe_put_cb(void *arg)
 {
@@ -396,10 +326,15 @@ pair1_pipe_get_cb(void *arg)
 		    ((hops = nni_msg_header_trim_u32(msg)) > 254)) {
 			BUMP_STAT(&s->stat_tx_malformed);
 			nni_msg_free(msg);
-			nni_msgq_aio_get(
-			    s->poly ? p->send_queue : s->uwq, &p->aio_get);
+			nni_msgq_aio_get(s->uwq, &p->aio_get);
 			return;
 		}
+#if NNG_TEST_LIB
+	} else if (s->inject_header) {
+		nni_aio_set_msg(&p->aio_send, msg);
+		nni_pipe_send(p->pipe, &p->aio_send);
+		return;
+#endif
 	} else {
 		// Strip off any previously existing header, such as when
 		// replying to messages.
@@ -429,9 +364,7 @@ pair1_pipe_send_cb(void *arg)
 		return;
 	}
 
-	// In polyamorous mode, we want to get from the send_queue; in
-	// monogamous we get from upper write queue.
-	nni_msgq_aio_get(s->poly ? p->send_queue : s->uwq, &p->aio_get);
+	nni_msgq_aio_get(s->uwq, &p->aio_get);
 }
 
 static void
@@ -443,8 +376,7 @@ pair1_sock_open(void *arg)
 static void
 pair1_sock_close(void *arg)
 {
-	pair1_sock *s = arg;
-	nni_aio_close(&s->aio_get);
+	NNI_ARG_UNUSED(arg);
 }
 
 static int
@@ -468,24 +400,19 @@ pair1_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
 	return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t));
 }
 
+
+#ifdef NNG_TEST_LIB
 static int
-pair1_sock_set_poly(void *arg, const void *buf, size_t sz, nni_opt_type t)
+pair1_set_test_inject_header(void *arg, const void *buf, size_t sz, nni_type t)
 {
 	pair1_sock *s = arg;
 	int         rv;
 	nni_mtx_lock(&s->mtx);
-	rv = s->started ? NNG_ESTATE : nni_copyin_bool(&s->poly, buf, sz, t);
-	nni_stat_set_value(&s->stat_poly, s->poly);
+	rv = nni_copyin_bool(&s->inject_header, buf, sz, t);
 	nni_mtx_unlock(&s->mtx);
 	return (rv);
 }
-
-static int
-pair1_sock_get_poly(void *arg, void *buf, size_t *szp, nni_opt_type t)
-{
-	pair1_sock *s = arg;
-	return (nni_copyout_bool(s->poly, buf, szp, t));
-}
+#endif
 
 static void
 pair1_sock_send(void *arg, nni_aio *aio)
@@ -519,11 +446,14 @@ static nni_option pair1_sock_options[] = {
 	    .o_get  = pair1_sock_get_max_ttl,
 	    .o_set  = pair1_sock_set_max_ttl,
 	},
+#ifdef NNG_TEST_LIB
 	{
-	    .o_name = NNG_OPT_PAIR1_POLY,
-	    .o_get  = pair1_sock_get_poly,
-	    .o_set  = pair1_sock_set_poly,
+	    // Test only option to pass header unmolested.  This allows
+	    // us to inject bad header contents.
+	    .o_name = "pair1_test_inject_header",
+	    .o_set  = pair1_set_test_inject_header,
 	},
+#endif
 	// terminate list
 	{
 	    .o_name = NULL,
diff --git a/src/protocol/pair1/pair1_poly.c b/src/protocol/pair1/pair1_poly.c
new file mode 100644
index 00000000..183c10da
--- /dev/null
+++ b/src/protocol/pair1/pair1_poly.c
@@ -0,0 +1,492 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2018 Capitar IT Group BV <info@capitar.com>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <stdlib.h>
+
+#include "core/nng_impl.h"
+#include "nng/protocol/pair1/pair.h"
+
+// Pair1 polyamorous mode.  The PAIRv1 protocol is normally a simple 1:1
+// messaging pattern, but this mode offers the ability to use a best-effort
+// multicast type of communication.  There are limitations however.
+// Most notably this does not interact well with nng_device type
+// proxies, and there is no support for raw mode.
+
+// THIS FEATURE IS DEPRECATED.  We discourage use in new applications.
+
+#define BUMP_STAT(x) nni_stat_inc_atomic(x, 1)
+
+typedef struct pair1poly_pipe pair1poly_pipe;
+typedef struct pair1poly_sock pair1poly_sock;
+
+static void pair1poly_sock_get_cb(void *);
+static void pair1poly_pipe_send_cb(void *);
+static void pair1poly_pipe_recv_cb(void *);
+static void pair1poly_pipe_get_cb(void *);
+static void pair1poly_pipe_put_cb(void *);
+static void pair1poly_pipe_fini(void *);
+
+// pair1poly_sock is our per-socket protocol private structure.
+struct pair1poly_sock {
+	nni_msgq *     uwq;
+	nni_msgq *     urq;
+	nni_sock *     sock;
+	nni_atomic_int ttl;
+	nni_mtx        mtx;
+	nni_idhash *   pipes;
+	nni_list       plist;
+	bool           started;
+	nni_aio        aio_get;
+	nni_stat_item  stat_poly;
+	nni_stat_item  stat_raw;
+	nni_stat_item  stat_reject_mismatch;
+	nni_stat_item  stat_reject_already;
+	nni_stat_item  stat_ttl_drop;
+	nni_stat_item  stat_rx_malformed;
+	nni_stat_item  stat_tx_malformed;
+	nni_stat_item  stat_tx_drop;
+};
+
+// pair1poly_pipe is our per-pipe protocol private structure.
+struct pair1poly_pipe {
+	nni_pipe *      pipe;
+	pair1poly_sock *pair;
+	nni_msgq *      send_queue;
+	nni_aio         aio_send;
+	nni_aio         aio_recv;
+	nni_aio         aio_get;
+	nni_aio         aio_put;
+	nni_list_node   node;
+};
+
+static void
+pair1poly_sock_fini(void *arg)
+{
+	pair1poly_sock *s = arg;
+
+	nni_aio_fini(&s->aio_get);
+	nni_idhash_fini(s->pipes);
+	nni_mtx_fini(&s->mtx);
+}
+
+static int
+pair1poly_sock_init(void *arg, nni_sock *sock)
+{
+	pair1poly_sock *s = arg;
+
+	if (nni_idhash_init(&s->pipes) != 0) {
+		return (NNG_ENOMEM);
+	}
+	NNI_LIST_INIT(&s->plist, pair1poly_pipe, node);
+
+	// Raw mode uses this.
+	nni_mtx_init(&s->mtx);
+
+	nni_aio_init(&s->aio_get, pair1poly_sock_get_cb, s);
+
+	nni_stat_init_bool(
+	    &s->stat_poly, "polyamorous", "polyamorous mode?", true);
+	nni_sock_add_stat(sock, &s->stat_poly);
+
+	nni_stat_init_bool(&s->stat_raw, "raw", "raw mode?", false);
+
+	nni_stat_init_atomic(&s->stat_reject_mismatch, "mismatch",
+	    "pipes rejected (protocol mismatch)");
+	nni_sock_add_stat(sock, &s->stat_reject_mismatch);
+
+	nni_stat_init_atomic(&s->stat_reject_already, "already",
+	    "pipes rejected (already connected)");
+	nni_sock_add_stat(sock, &s->stat_reject_already);
+
+	nni_stat_init_atomic(&s->stat_ttl_drop, "ttl_drop",
+	    "messages dropped due to too many hops");
+	nni_stat_set_unit(&s->stat_ttl_drop, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_ttl_drop);
+
+	// This can only increment in polyamorous mode.
+	nni_stat_init_atomic(
+	    &s->stat_tx_drop, "tx_drop", "messages dropped undeliverable");
+	nni_stat_set_unit(&s->stat_tx_drop, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_tx_drop);
+
+	nni_stat_init_atomic(&s->stat_rx_malformed, "rx_malformed",
+	    "malformed messages received");
+	nni_stat_set_unit(&s->stat_rx_malformed, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_rx_malformed);
+
+	nni_stat_init_atomic(&s->stat_tx_malformed, "tx_malformed",
+	    "malformed messages not sent");
+	nni_stat_set_unit(&s->stat_tx_malformed, NNG_UNIT_MESSAGES);
+
+	s->sock = sock;
+	s->uwq  = nni_sock_sendq(sock);
+	s->urq  = nni_sock_recvq(sock);
+	nni_atomic_init(&s->ttl);
+	nni_atomic_set(&s->ttl, 8);
+
+	return (0);
+}
+
+static void
+pair1poly_pipe_stop(void *arg)
+{
+	pair1poly_pipe *p = arg;
+
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+	nni_aio_stop(&p->aio_put);
+	nni_aio_stop(&p->aio_get);
+}
+
+static void
+pair1poly_pipe_fini(void *arg)
+{
+	pair1poly_pipe *p = arg;
+
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+	nni_aio_fini(&p->aio_put);
+	nni_aio_fini(&p->aio_get);
+	nni_msgq_fini(p->send_queue);
+}
+
+static int
+pair1poly_pipe_init(void *arg, nni_pipe *pipe, void *pair)
+{
+	pair1poly_pipe *p = arg;
+	int             rv;
+
+	nni_aio_init(&p->aio_send, pair1poly_pipe_send_cb, p);
+	nni_aio_init(&p->aio_recv, pair1poly_pipe_recv_cb, p);
+	nni_aio_init(&p->aio_get, pair1poly_pipe_get_cb, p);
+	nni_aio_init(&p->aio_put, pair1poly_pipe_put_cb, p);
+
+	if ((rv = nni_msgq_init(&p->send_queue, 2)) != 0) {
+		pair1poly_pipe_fini(p);
+		return (rv);
+	}
+
+	p->pipe = pipe;
+	p->pair = pair;
+
+	return (0);
+}
+
+static int
+pair1poly_pipe_start(void *arg)
+{
+	pair1poly_pipe *p = arg;
+	pair1poly_sock *s = p->pair;
+	uint32_t        id;
+	int             rv;
+
+	nni_mtx_lock(&s->mtx);
+	if (nni_pipe_peer(p->pipe) != NNG_PAIR1_PEER) {
+		nni_mtx_unlock(&s->mtx);
+		BUMP_STAT(&s->stat_reject_mismatch);
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
+	id = nni_pipe_id(p->pipe);
+	if ((rv = nni_idhash_insert(s->pipes, id, p)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		return (rv);
+	}
+	if (!s->started) {
+		nni_msgq_aio_get(s->uwq, &s->aio_get);
+	}
+	nni_list_append(&s->plist, p);
+	s->started = true;
+	nni_mtx_unlock(&s->mtx);
+
+	// Schedule a get.  In polyamorous mode we get on the per pipe
+	// send_queue, as the socket distributes to us. In monogamous mode
+	// we bypass and get from the upper write queue directly (saving a
+	// set of context switches).
+	nni_msgq_aio_get(p->send_queue, &p->aio_get);
+
+	// And the pipe read of course.
+	nni_pipe_recv(p->pipe, &p->aio_recv);
+
+	return (0);
+}
+
+static void
+pair1poly_pipe_close(void *arg)
+{
+	pair1poly_pipe *p = arg;
+	pair1poly_sock *s = p->pair;
+
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+	nni_aio_close(&p->aio_put);
+	nni_aio_close(&p->aio_get);
+
+	nni_mtx_lock(&s->mtx);
+	nni_idhash_remove(s->pipes, nni_pipe_id(p->pipe));
+	nni_list_node_remove(&p->node);
+	nni_mtx_unlock(&s->mtx);
+
+	nni_msgq_close(p->send_queue);
+}
+
+static void
+pair1poly_pipe_recv_cb(void *arg)
+{
+	pair1poly_pipe *p = arg;
+	pair1poly_sock *s = p->pair;
+	nni_msg *       msg;
+	uint32_t        hdr;
+	nni_pipe *      pipe = p->pipe;
+	size_t          len;
+
+	if (nni_aio_result(&p->aio_recv) != 0) {
+		nni_pipe_close(p->pipe);
+		return;
+	}
+
+	msg = nni_aio_get_msg(&p->aio_recv);
+	nni_aio_set_msg(&p->aio_recv, NULL);
+
+	// Store the pipe ID.
+	nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
+
+	// If the message is missing the hop count header, scrap it.
+	if ((nni_msg_len(msg) < sizeof(uint32_t)) ||
+	    ((hdr = nni_msg_trim_u32(msg)) > 0xff)) {
+		BUMP_STAT(&s->stat_rx_malformed);
+		nni_msg_free(msg);
+		nni_pipe_close(pipe);
+		return;
+	}
+
+	len = nni_msg_len(msg);
+
+	// If we bounced too many times, discard the message, but
+	// keep getting more.
+	if ((int) hdr > nni_atomic_get(&s->ttl)) {
+		BUMP_STAT(&s->stat_ttl_drop);
+		nni_msg_free(msg);
+		nni_pipe_recv(pipe, &p->aio_recv);
+		return;
+	}
+
+	// Store the hop count in the header.
+	nni_msg_header_must_append_u32(msg, hdr);
+
+	// Send the message up.
+	nni_aio_set_msg(&p->aio_put, msg);
+	nni_sock_bump_rx(s->sock, len);
+	nni_msgq_aio_put(s->urq, &p->aio_put);
+}
+
+static void
+pair1poly_sock_get_cb(void *arg)
+{
+	pair1poly_pipe *p;
+	pair1poly_sock *s = arg;
+	nni_msg *       msg;
+	uint32_t        id;
+
+	if (nni_aio_result(&s->aio_get) != 0) {
+		// Socket closing...
+		return;
+	}
+
+	msg = nni_aio_get_msg(&s->aio_get);
+	nni_aio_set_msg(&s->aio_get, NULL);
+
+	p = NULL;
+	nni_mtx_lock(&s->mtx);
+	// If no pipe was requested, we look for any connected peer.
+	if (((id = nni_msg_get_pipe(msg)) == 0) &&
+	    (!nni_list_empty(&s->plist))) {
+		p = nni_list_first(&s->plist);
+	} else {
+		nni_idhash_find(s->pipes, id, (void **) &p);
+	}
+
+	// Try a non-blocking send.  If this fails we just discard the
+	// message.  We have to do this to avoid head-of-line blocking
+	// for messages sent to other pipes.  Note that there is some
+	// buffering in the send_queue.
+	if ((p == NULL) || nni_msgq_tryput(p->send_queue, msg) != 0) {
+		BUMP_STAT(&s->stat_tx_drop);
+		nni_msg_free(msg);
+	}
+
+	nni_mtx_unlock(&s->mtx);
+	nni_msgq_aio_get(s->uwq, &s->aio_get);
+}
+
+static void
+pair1poly_pipe_put_cb(void *arg)
+{
+	pair1poly_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_put) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_put));
+		nni_aio_set_msg(&p->aio_put, NULL);
+		nni_pipe_close(p->pipe);
+		return;
+	}
+	nni_pipe_recv(p->pipe, &p->aio_recv);
+}
+
+static void
+pair1poly_pipe_get_cb(void *arg)
+{
+	pair1poly_pipe *p = arg;
+	nni_msg *       msg;
+
+	if (nni_aio_result(&p->aio_get) != 0) {
+		nni_pipe_close(p->pipe);
+		return;
+	}
+
+	msg = nni_aio_get_msg(&p->aio_get);
+	nni_aio_set_msg(&p->aio_get, NULL);
+
+	// Cooked mode messages have no header so we have to add one.
+	// Strip off any previously existing header, such as when
+	// replying to messages.
+	nni_msg_header_clear(msg);
+
+	// Insert the hops header.
+	nni_msg_header_must_append_u32(msg, 1);
+
+	nni_aio_set_msg(&p->aio_send, msg);
+	nni_pipe_send(p->pipe, &p->aio_send);
+}
+
+static void
+pair1poly_pipe_send_cb(void *arg)
+{
+	pair1poly_pipe *p = arg;
+
+	if (nni_aio_result(&p->aio_send) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_send));
+		nni_aio_set_msg(&p->aio_send, NULL);
+		nni_pipe_close(p->pipe);
+		return;
+	}
+
+	nni_msgq_aio_get(p->send_queue, &p->aio_get);
+}
+
+static void
+pair1poly_sock_open(void *arg)
+{
+	NNI_ARG_UNUSED(arg);
+}
+
+static void
+pair1poly_sock_close(void *arg)
+{
+	pair1poly_sock *s = arg;
+	nni_aio_close(&s->aio_get);
+}
+
+static int
+pair1poly_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+{
+	pair1poly_sock *s = arg;
+	int             rv;
+	int             ttl;
+
+	if ((rv = nni_copyin_int(&ttl, buf, sz, 1, NNI_MAX_MAX_TTL, t)) == 0) {
+		nni_atomic_set(&s->ttl, ttl);
+	}
+
+	return (rv);
+}
+
+static int
+pair1poly_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	pair1poly_sock *s = arg;
+	return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t));
+}
+
+static int
+pair1poly_get_poly(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	NNI_ARG_UNUSED(arg);
+	return (nni_copyout_bool(true, buf, szp, t));
+}
+
+static void
+pair1poly_sock_send(void *arg, nni_aio *aio)
+{
+	pair1poly_sock *s = arg;
+
+	nni_sock_bump_tx(s->sock, nni_msg_len(nni_aio_get_msg(aio)));
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+pair1poly_sock_recv(void *arg, nni_aio *aio)
+{
+	pair1poly_sock *s = arg;
+
+	nni_msgq_aio_get(s->urq, aio);
+}
+
+static nni_proto_pipe_ops pair1poly_pipe_ops = {
+	.pipe_size  = sizeof(pair1poly_pipe),
+	.pipe_init  = pair1poly_pipe_init,
+	.pipe_fini  = pair1poly_pipe_fini,
+	.pipe_start = pair1poly_pipe_start,
+	.pipe_close = pair1poly_pipe_close,
+	.pipe_stop  = pair1poly_pipe_stop,
+};
+
+static nni_option pair1poly_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_MAXTTL,
+	    .o_get  = pair1poly_get_max_ttl,
+	    .o_set  = pair1poly_set_max_ttl,
+	},
+	{
+	    .o_name = NNG_OPT_PAIR1_POLY,
+	    .o_get  = pair1poly_get_poly,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
+static nni_proto_sock_ops pair1poly_sock_ops = {
+	.sock_size    = sizeof(pair1poly_sock),
+	.sock_init    = pair1poly_sock_init,
+	.sock_fini    = pair1poly_sock_fini,
+	.sock_open    = pair1poly_sock_open,
+	.sock_close   = pair1poly_sock_close,
+	.sock_recv    = pair1poly_sock_recv,
+	.sock_send    = pair1poly_sock_send,
+	.sock_options = pair1poly_sock_options,
+};
+
+static nni_proto pair1poly_proto = {
+	.proto_version  = NNI_PROTOCOL_VERSION,
+	.proto_self     = { NNG_PAIR1_SELF, NNG_PAIR1_SELF_NAME },
+	.proto_peer     = { NNG_PAIR1_PEER, NNG_PAIR1_PEER_NAME },
+	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
+	.proto_sock_ops = &pair1poly_sock_ops,
+	.proto_pipe_ops = &pair1poly_pipe_ops,
+};
+
+int
+nng_pair1_open_poly(nng_socket *sock)
+{
+	return (nni_proto_open(sock, &pair1poly_proto));
+}


Test patch:
diff --git a/src/protocol/pair1/pair1_poly_test.c b/src/protocol/pair1/pair1_poly_test.c
new file mode 100644
index 00000000..c4acbd86
--- /dev/null
+++ b/src/protocol/pair1/pair1_poly_test.c
@@ -0,0 +1,357 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2017 Capitar IT Group BV <info@capitar.com>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <string.h>
+
+#include <nng/nng.h>
+#include <nng/protocol/pair0/pair.h>
+#include <nng/protocol/pair1/pair.h>
+
+#include <testutil.h>
+
+#include <acutest.h>
+
+#define SECOND 1000
+
+#define APPEND_STR(m, s) TEST_NNG_PASS(nng_msg_append(m, s, strlen(s)))
+#define CHECK_STR(m, s)                          \
+	TEST_CHECK(nng_msg_len(m) == strlen(s)); \
+	TEST_CHECK(memcmp(nng_msg_body(m), s, strlen(s)) == 0)
+
+void
+test_poly_best_effort(void)
+{
+	nng_socket s1;
+	nng_socket c1;
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s1));
+	TEST_NNG_PASS(nng_pair1_open(&c1));
+
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_RECVBUF, 1));
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_SENDBUF, 1));
+	TEST_NNG_PASS(nng_setopt_int(c1, NNG_OPT_RECVBUF, 1));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND));
+
+	TEST_NNG_PASS(testutil_marry(s1, c1));
+
+	for (int i = 0; i < 10; i++) {
+		TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+		TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	}
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c1));
+}
+
+void
+test_poly_cooked(void)
+{
+	nng_socket s1;
+	nng_socket c1;
+	nng_socket c2;
+	nng_msg *  msg;
+	bool       v;
+	nng_pipe   p1;
+	nng_pipe   p2;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s1));
+	TEST_NNG_PASS(nng_pair1_open(&c1));
+	TEST_NNG_PASS(nng_pair1_open(&c2));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_SENDTIMEO, SECOND));
+	TEST_NNG_PASS(nng_setopt_ms(c2, NNG_OPT_SENDTIMEO, SECOND));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 10));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 10));
+	TEST_NNG_PASS(nng_setopt_ms(c2, NNG_OPT_RECVTIMEO, SECOND / 10));
+
+	TEST_NNG_PASS(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v));
+	TEST_CHECK(v == true);
+
+	TEST_NNG_PASS(testutil_marry(s1, c1));
+	TEST_NNG_PASS(testutil_marry(s1, c2));
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	APPEND_STR(msg, "ONE");
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	CHECK_STR(msg, "ONE");
+	p1 = nng_msg_get_pipe(msg);
+	TEST_CHECK(nng_pipe_id(p1) > 0);
+	nng_msg_free(msg);
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	APPEND_STR(msg, "TWO");
+	TEST_NNG_PASS(nng_sendmsg(c2, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	CHECK_STR(msg, "TWO");
+	p2 = nng_msg_get_pipe(msg);
+	TEST_CHECK(nng_pipe_id(p2) > 0);
+	nng_msg_free(msg);
+
+	TEST_CHECK(nng_pipe_id(p1) != nng_pipe_id(p2));
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+
+	nng_msg_set_pipe(msg, p1);
+	APPEND_STR(msg, "UNO");
+	TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(c1, &msg, 0));
+	CHECK_STR(msg, "UNO");
+	nng_msg_free(msg);
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	nng_msg_set_pipe(msg, p2);
+	APPEND_STR(msg, "DOS");
+	TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(c2, &msg, 0));
+	CHECK_STR(msg, "DOS");
+	nng_msg_free(msg);
+
+	TEST_NNG_PASS(nng_close(c1));
+
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	nng_msg_set_pipe(msg, p1);
+	APPEND_STR(msg, "EIN");
+	TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(c2, &msg, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c2));
+}
+
+void
+test_poly_default(void)
+{
+	nng_socket s1;
+	nng_socket c1;
+	nng_socket c2;
+	nng_msg *  msg;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s1));
+	TEST_NNG_PASS(nng_pair1_open(&c1));
+	TEST_NNG_PASS(nng_pair1_open(&c2));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_SENDTIMEO, SECOND));
+	TEST_NNG_PASS(nng_setopt_ms(c2, NNG_OPT_SENDTIMEO, SECOND));
+
+	TEST_NNG_PASS(testutil_marry(s1, c1));
+	TEST_NNG_PASS(testutil_marry(s1, c2));
+
+	// This assumes poly picks the first suitor.  Applications
+	// should not make the same assumption.
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	APPEND_STR(msg, "YES");
+	TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(c1, &msg, 0));
+	CHECK_STR(msg, "YES");
+	nng_msg_free(msg);
+
+	TEST_NNG_PASS(nng_close(c1));
+	testutil_sleep(10);
+
+	// Verify that the other pipe is chosen as the next suitor.
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	APPEND_STR(msg, "AGAIN");
+	TEST_NNG_PASS(nng_sendmsg(s1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(c2, &msg, 0));
+	CHECK_STR(msg, "AGAIN");
+	nng_msg_free(msg);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c2));
+}
+
+void
+test_poly_close_abort(void)
+{
+	nng_socket s;
+	nng_socket c;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s));
+	TEST_NNG_PASS(nng_pair1_open(&c));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_SENDTIMEO, 200));
+	TEST_NNG_PASS(nng_setopt_int(s, NNG_OPT_RECVBUF, 1));
+	TEST_NNG_PASS(nng_setopt_int(c, NNG_OPT_SENDBUF, 20));
+
+	TEST_NNG_PASS(testutil_marry(c, s));
+
+	for (int i = 0; i < 20; i++) {
+		TEST_NNG_SEND_STR(c, "TEST");
+	}
+	testutil_sleep(50);
+
+	TEST_NNG_PASS(nng_close(s));
+	TEST_NNG_PASS(nng_close(c));
+}
+
+
+void
+test_poly_recv_no_header(void)
+{
+	nng_socket s;
+	nng_socket c;
+	nng_msg *  m;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s));
+	TEST_NNG_PASS(nng_pair1_open(&c));
+	TEST_NNG_PASS(nng_setopt_bool(c, "pair1_test_inject_header", true));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_SENDTIMEO, 200));
+
+	TEST_NNG_PASS(testutil_marry(c, s));
+
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	TEST_NNG_PASS(nng_sendmsg(c, m, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s, &m, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(c));
+	TEST_NNG_PASS(nng_close(s));
+}
+
+void
+test_poly_recv_garbage(void)
+{
+	nng_socket s;
+	nng_socket c;
+	nng_msg *  m;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s));
+	TEST_NNG_PASS(nng_pair1_open(&c));
+	TEST_NNG_PASS(nng_setopt_bool(c, "pair1_test_inject_header", true));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_SENDTIMEO, 200));
+
+	TEST_NNG_PASS(testutil_marry(c, s));
+
+	// ridiculous hop count
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(m, 0x1000));
+	TEST_NNG_PASS(nng_sendmsg(c, m, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s, &m, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(c));
+	TEST_NNG_PASS(nng_close(s));
+}
+
+void
+test_poly_ttl(void)
+{
+	nng_socket s1;
+	nng_socket c1;
+	nng_msg *  msg;
+	uint32_t   val;
+	int        ttl;
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s1));
+	TEST_NNG_PASS(nng_pair1_open_raw(&c1));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5));
+
+	// cannot set insane TTLs
+	TEST_NNG_FAIL(nng_setopt_int(s1, NNG_OPT_MAXTTL, 0), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_int(s1, NNG_OPT_MAXTTL, 1000), NNG_EINVAL);
+	ttl = 8;
+	TEST_NNG_FAIL(nng_setopt(s1, NNG_OPT_MAXTTL, &ttl, 1), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_bool(s1, NNG_OPT_MAXTTL, true), NNG_EBADTYPE);
+
+	TEST_NNG_PASS(testutil_marry(s1, c1));
+
+	// Let's check enforcement of TTL
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 4));
+	TEST_NNG_PASS(nng_getopt_int(s1, NNG_OPT_MAXTTL, &ttl));
+	TEST_CHECK(ttl == 4);
+
+	// Bad TTL bounces
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 4));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
+
+	// Good TTL passes
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 0xFEEDFACE));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 3));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &val));
+	TEST_CHECK(val == 0xFEEDFACE);
+	TEST_NNG_PASS(nng_msg_header_trim_u32(msg, &val));
+	TEST_CHECK(val == 4);
+	nng_msg_free(msg);
+
+	// Large TTL passes
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 1234));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 14));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &val));
+	TEST_CHECK(val == 1234);
+	TEST_NNG_PASS(nng_msg_header_trim_u32(msg, &val));
+	TEST_CHECK(val == 15);
+	nng_msg_free(msg);
+
+	// Max TTL fails
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 15));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c1));
+}
+
+void
+test_poly_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char       addr[64];
+
+	testutil_scratch_addr("inproc", sizeof(addr), addr);
+
+	TEST_NNG_PASS(nng_pair1_open_poly(&s1));
+	TEST_NNG_PASS(nng_pair0_open(&s2));
+
+	TEST_NNG_PASS(nng_listen(s1, addr, NULL, 0));
+	TEST_NNG_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	testutil_sleep(100);
+	TEST_NNG_PASS(nng_stats_get(&stats));
+
+	TEST_CHECK(stats != NULL);
+	TEST_CHECK((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	TEST_CHECK((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	TEST_CHECK(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	TEST_CHECK(nng_stat_value(reject) > 0);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(s2));
+	nng_stats_free(stats);
+}
+
+TEST_LIST = {
+	{ "pair1 polyamorous best effort", test_poly_best_effort },
+	{ "pair1 polyamorous cooked", test_poly_cooked },
+	{ "pair1 polyamorous default", test_poly_default },
+	{ "pair1 polyamorous recv no header", test_poly_recv_no_header },
+	{ "pair1 polyamorous recv garbage", test_poly_recv_garbage },
+	{ "pair1 polyamorous ttl", test_poly_ttl },
+	{ "pair1 polyamorous close abort", test_poly_close_abort },
+	{ "pair1 polyamorous validate peer", test_poly_validate_peer },
+
+	{ NULL, NULL },
+};
diff --git a/src/protocol/pair1/pair1_test.c b/src/protocol/pair1/pair1_test.c
index 7011606a..0f0131b5 100644
--- a/src/protocol/pair1/pair1_test.c
+++ b/src/protocol/pair1/pair1_test.c
@@ -11,8 +11,8 @@
 #include <string.h>
 
 #include <nng/nng.h>
-#include <nng/protocol/pair1/pair.h>
 #include <nng/protocol/pair0/pair.h>
+#include <nng/protocol/pair1/pair.h>
 
 #include <testutil.h>
 
@@ -188,33 +188,33 @@ test_mono_raw_header(void)
 	nng_msg *  msg;
 	uint32_t   v;
 
-	TEST_CHECK(nng_pair1_open_raw(&s1) == 0);
-	TEST_CHECK(nng_pair1_open_raw(&c1) == 0);
+	TEST_NNG_PASS(nng_pair1_open_raw(&s1));
+	TEST_NNG_PASS(nng_pair1_open_raw(&c1));
 
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5));
+	TEST_NNG_PASS(testutil_marry(s1, c1));
 
 	// Missing bits in the header
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
 
 	// Valid header works
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_append_u32(msg, 0xFEEDFACE) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 1) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	TEST_CHECK(nng_msg_trim_u32(msg, &v) == 0);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 0xFEEDFACE));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 1));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &v));
 	TEST_CHECK(v == 0xFEEDFACE);
 	nng_msg_free(msg);
 
 	// Header with reserved bits set dropped
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 0xDEAD0000) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 0xDEAD0000));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
 
 	// Header with no chance to add another hop gets dropped
 	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
@@ -223,313 +223,40 @@ test_mono_raw_header(void)
 	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
 
 	// With the same bits clear it works
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_append_u32(msg, 0xFEEDFACE) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 1) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	TEST_CHECK(nng_msg_trim_u32(msg, &v) == 0);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 0xFEEDFACE));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 1));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &v));
 	TEST_CHECK(v == 0xFEEDFACE);
 	nng_msg_free(msg);
 
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c1) == 0);
-}
-
-void
-test_poly_best_effort(void)
-{
-	nng_socket s1;
-	nng_socket c1;
-	nng_msg *  msg;
-
-	TEST_CHECK(nng_pair1_open(&s1) == 0);
-	TEST_CHECK(nng_pair1_open(&c1) == 0);
-
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == 0);
-
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_RECVBUF, 1) == 0);
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_SENDBUF, 1) == 0);
-	TEST_CHECK(nng_setopt_int(c1, NNG_OPT_RECVBUF, 1) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND) == 0);
-
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
-
-	for (int i = 0; i < 10; i++) {
-		TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-		TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	}
-
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c1) == 0);
-}
-
-void
-test_poly_cooked(void)
-{
-	nng_socket s1;
-	nng_socket c1;
-	nng_socket c2;
-	nng_msg *  msg;
-	bool       v;
-	nng_pipe   p1;
-	nng_pipe   p2;
-
-	TEST_CHECK(nng_pair1_open(&s1) == 0);
-	TEST_CHECK(nng_pair1_open(&c1) == 0);
-	TEST_CHECK(nng_pair1_open(&c2) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(c2, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 10) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 10) == 0);
-	TEST_CHECK(nng_setopt_ms(c2, NNG_OPT_RECVTIMEO, SECOND / 10) == 0);
-
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == false);
-
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == 0);
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == true);
-
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
-	TEST_CHECK(testutil_marry(s1, c2) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "ONE");
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	CHECK_STR(msg, "ONE");
-	p1 = nng_msg_get_pipe(msg);
-	TEST_CHECK(nng_pipe_id(p1) > 0);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "TWO");
-	TEST_CHECK(nng_sendmsg(c2, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	CHECK_STR(msg, "TWO");
-	p2 = nng_msg_get_pipe(msg);
-	TEST_CHECK(nng_pipe_id(p2) > 0);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_pipe_id(p1) != nng_pipe_id(p2));
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-
-	nng_msg_set_pipe(msg, p1);
-	APPEND_STR(msg, "UNO");
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c1, &msg, 0) == 0);
-	CHECK_STR(msg, "UNO");
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	nng_msg_set_pipe(msg, p2);
-	APPEND_STR(msg, "DOS");
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c2, &msg, 0) == 0);
-	CHECK_STR(msg, "DOS");
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_close(c1) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	nng_msg_set_pipe(msg, p1);
-	APPEND_STR(msg, "EIN");
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c2, &msg, 0) == NNG_ETIMEDOUT);
-
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c2) == 0);
-}
-
-void
-test_poly_late(void)
-{
-	nng_socket s1;
-	nng_socket c1;
-	bool       v;
-
-	TEST_CHECK(nng_pair1_open(&s1) == 0);
-	TEST_CHECK(nng_pair1_open(&c1) == 0);
-
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == false);
-
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == 0);
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == true);
-
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
-
-	TEST_CHECK(
-	    nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == NNG_ESTATE);
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c1) == 0);
-}
-
-void
-test_poly_default(void)
-{
-	nng_socket s1;
-	nng_socket c1;
-	nng_socket c2;
-	nng_msg *  msg;
-
-	TEST_CHECK(nng_pair1_open(&s1) == 0);
-	TEST_CHECK(nng_pair1_open(&c1) == 0);
-	TEST_CHECK(nng_pair1_open(&c2) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_SENDTIMEO, SECOND) == 0);
-	TEST_CHECK(nng_setopt_ms(c2, NNG_OPT_SENDTIMEO, SECOND) == 0);
-
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == 0);
-
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
-	TEST_CHECK(testutil_marry(s1, c2) == 0);
-
-	// This assumes poly picks the first suitor.  Applications
-	// should not make the same assumption.
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "YES");
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c1, &msg, 0) == 0);
-	CHECK_STR(msg, "YES");
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_close(c1) == 0);
-	testutil_sleep(10);
-
-	// Verify that the other pipe is chosen as the next suitor.
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "AGAIN");
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c2, &msg, 0) == 0);
-	CHECK_STR(msg, "AGAIN");
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c2) == 0);
-}
-
-void
-test_poly_raw(void)
-{
-	nng_socket s1;
-	nng_socket c1;
-	nng_socket c2;
-	nng_msg *  msg;
-	bool       v;
-	uint32_t   hops;
-	nng_pipe   p1;
-	nng_pipe   p2;
-
-	TEST_CHECK(nng_pair1_open_raw(&s1) == 0);
-	TEST_CHECK(nng_pair1_open(&c1) == 0);
-	TEST_CHECK(nng_pair1_open(&c2) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-	TEST_CHECK(nng_setopt_ms(c2, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == 0);
-
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_PAIR1_POLY, true) == 0);
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_PAIR1_POLY, &v) == 0);
-	TEST_CHECK(v == true);
-
-	v = false;
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_RAW, &v) == 0);
-	TEST_CHECK(v == true);
-
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
-	TEST_CHECK(testutil_marry(s1, c2) == 0);
-
-	// send/recv works
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "ONE");
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	CHECK_STR(msg, "ONE");
-	p1 = nng_msg_get_pipe(msg);
-	TEST_CHECK(nng_pipe_id(p1) > 0);
-	TEST_CHECK(nng_msg_header_trim_u32(msg, &hops) == 0);
-	TEST_CHECK(hops == 1);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "TWO");
-	TEST_CHECK(nng_sendmsg(c2, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	CHECK_STR(msg, "TWO");
-	p2 = nng_msg_get_pipe(msg);
-	TEST_CHECK(nng_pipe_id(p2) > 0);
-	TEST_CHECK(nng_msg_header_trim_u32(msg, &hops) == 0);
-	TEST_CHECK(hops == 1);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_pipe_id(p1) != nng_pipe_id(p2));
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	nng_msg_set_pipe(msg, p1);
-	APPEND_STR(msg, "UNO");
-	TEST_CHECK(nng_msg_header_append_u32(msg, 1) == 0);
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c1, &msg, 0) == 0);
-	CHECK_STR(msg, "UNO");
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	nng_msg_set_pipe(msg, p2);
-	APPEND_STR(msg, "DOS");
-	TEST_CHECK(nng_msg_header_append_u32(msg, 1) == 0);
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c2, &msg, 0) == 0);
-	CHECK_STR(msg, "DOS");
-	nng_msg_free(msg);
-
-	// Verify closing the pipe stops any of its traffic
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	APPEND_STR(msg, "ONE");
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	CHECK_STR(msg, "ONE");
-	p1 = nng_msg_get_pipe(msg);
-	TEST_CHECK(nng_pipe_id(p1) > 0);
-	nng_msg_free(msg);
-
-	TEST_CHECK(nng_close(c1) == 0);
-
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	nng_msg_set_pipe(msg, p1);
-	APPEND_STR(msg, "EIN");
-	TEST_CHECK(nng_msg_header_append_u32(msg, 1) == 0);
-	TEST_CHECK(nng_sendmsg(s1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(c2, &msg, 0) == NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c1));
 }
 
 void
-test_raw(void)
+test_pair1_raw(void)
 {
 	nng_socket s1;
 	bool       raw;
 
-	TEST_CHECK(nng_pair1_open(&s1) == 0);
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_RAW, &raw) == 0);
+	TEST_NNG_PASS(nng_pair1_open(&s1));
+	TEST_NNG_PASS(nng_getopt_bool(s1, NNG_OPT_RAW, &raw));
 	TEST_CHECK(raw == false);
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_RAW, true) == NNG_EREADONLY);
-	TEST_CHECK(nng_close(s1) == 0);
+	TEST_NNG_FAIL(nng_setopt_bool(s1, NNG_OPT_RAW, true), NNG_EREADONLY);
+	TEST_NNG_PASS(nng_close(s1));
 
-	TEST_CHECK(nng_pair1_open_raw(&s1) == 0);
-	TEST_CHECK(nng_getopt_bool(s1, NNG_OPT_RAW, &raw) == 0);
+	TEST_NNG_PASS(nng_pair1_open_raw(&s1));
+	TEST_NNG_PASS(nng_getopt_bool(s1, NNG_OPT_RAW, &raw));
 	TEST_CHECK(raw == true);
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_RAW, false) == NNG_EREADONLY);
-	TEST_CHECK(nng_close(s1) == 0);
+	TEST_NNG_FAIL(nng_setopt_bool(s1, NNG_OPT_RAW, false), NNG_EREADONLY);
+	TEST_NNG_PASS(nng_close(s1));
 }
 
 void
-test_ttl(void)
+test_pair1_ttl(void)
 {
 	nng_socket s1;
 	nng_socket c1;
@@ -537,69 +264,69 @@ test_ttl(void)
 	uint32_t   val;
 	int        ttl;
 
-	TEST_CHECK(nng_pair1_open_raw(&s1) == 0);
-	TEST_CHECK(nng_pair1_open_raw(&c1) == 0);
-	TEST_CHECK(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
-	TEST_CHECK(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5) == 0);
+	TEST_NNG_PASS(nng_pair1_open_raw(&s1));
+	TEST_NNG_PASS(nng_pair1_open_raw(&c1));
+	TEST_NNG_PASS(nng_setopt_ms(s1, NNG_OPT_RECVTIMEO, SECOND / 5));
+	TEST_NNG_PASS(nng_setopt_ms(c1, NNG_OPT_RECVTIMEO, SECOND / 5));
 
 	// cannot set insane TTLs
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_MAXTTL, 0) == NNG_EINVAL);
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_MAXTTL, 1000) == NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_int(s1, NNG_OPT_MAXTTL, 0), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_int(s1, NNG_OPT_MAXTTL, 1000), NNG_EINVAL);
 	ttl = 8;
-	TEST_CHECK(nng_setopt(s1, NNG_OPT_MAXTTL, &ttl, 1) == NNG_EINVAL);
-	TEST_CHECK(nng_setopt_bool(s1, NNG_OPT_MAXTTL, true) == NNG_EBADTYPE);
+	TEST_NNG_FAIL(nng_setopt(s1, NNG_OPT_MAXTTL, &ttl, 1), NNG_EINVAL);
+	TEST_NNG_FAIL(nng_setopt_bool(s1, NNG_OPT_MAXTTL, true), NNG_EBADTYPE);
 
-	TEST_CHECK(testutil_marry(s1, c1) == 0);
+	TEST_NNG_PASS(testutil_marry(s1, c1));
 
 	// Let's check enforcement of TTL
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_MAXTTL, 4) == 0);
-	TEST_CHECK(nng_getopt_int(s1, NNG_OPT_MAXTTL, &ttl) == 0);
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 4));
+	TEST_NNG_PASS(nng_getopt_int(s1, NNG_OPT_MAXTTL, &ttl));
 	TEST_CHECK(ttl == 4);
 
 	// Bad TTL bounces
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 4) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == NNG_ETIMEDOUT);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 4));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
 
 	// Good TTL passes
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_append_u32(msg, 0xFEEDFACE) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 3) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	TEST_CHECK(nng_msg_trim_u32(msg, &val) == 0);
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 0xFEEDFACE));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 3));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &val));
 	TEST_CHECK(val == 0xFEEDFACE);
-	TEST_CHECK(nng_msg_header_trim_u32(msg, &val) == 0);
+	TEST_NNG_PASS(nng_msg_header_trim_u32(msg, &val));
 	TEST_CHECK(val == 4);
 	nng_msg_free(msg);
 
 	// Large TTL passes
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15) == 0);
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_append_u32(msg, 1234) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 14) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == 0);
-	TEST_CHECK(nng_msg_trim_u32(msg, &val) == 0);
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(msg, 1234));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 14));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_PASS(nng_recvmsg(s1, &msg, 0));
+	TEST_NNG_PASS(nng_msg_trim_u32(msg, &val));
 	TEST_CHECK(val == 1234);
-	TEST_CHECK(nng_msg_header_trim_u32(msg, &val) == 0);
+	TEST_NNG_PASS(nng_msg_header_trim_u32(msg, &val));
 	TEST_CHECK(val == 15);
 	nng_msg_free(msg);
 
 	// Max TTL fails
-	TEST_CHECK(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15) == 0);
-	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
-	TEST_CHECK(nng_msg_header_append_u32(msg, 15) == 0);
-	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
-	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == NNG_ETIMEDOUT);
-
-	TEST_CHECK(nng_close(s1) == 0);
-	TEST_CHECK(nng_close(c1) == 0);
+	TEST_NNG_PASS(nng_setopt_int(s1, NNG_OPT_MAXTTL, 15));
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 15));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(s1));
+	TEST_NNG_PASS(nng_close(c1));
 }
 
 void
-test_validate_peer(void)
+test_pair1_validate_peer(void)
 {
 	nng_socket s1, s2;
 	nng_stat * stats;
@@ -629,20 +356,65 @@ test_validate_peer(void)
 	nng_stats_free(stats);
 }
 
+void
+test_pair1_recv_no_header(void)
+{
+	nng_socket s;
+	nng_socket c;
+	nng_msg *  m;
+
+	TEST_NNG_PASS(nng_pair1_open(&s));
+	TEST_NNG_PASS(nng_pair1_open(&c));
+	TEST_NNG_PASS(nng_setopt_bool(c, "pair1_test_inject_header", true));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_SENDTIMEO, 200));
+
+	TEST_NNG_PASS(testutil_marry(c, s));
+
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	TEST_NNG_PASS(nng_sendmsg(c, m, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s, &m, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(c));
+	TEST_NNG_PASS(nng_close(s));
+}
+
+void
+test_pair1_recv_garbage(void)
+{
+	nng_socket s;
+	nng_socket c;
+	nng_msg *  m;
+
+	TEST_NNG_PASS(nng_pair1_open(&s));
+	TEST_NNG_PASS(nng_pair1_open(&c));
+	TEST_NNG_PASS(nng_setopt_bool(c, "pair1_test_inject_header", true));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_RECVTIMEO, 100));
+	TEST_NNG_PASS(nng_setopt_ms(s, NNG_OPT_SENDTIMEO, 200));
+
+	TEST_NNG_PASS(testutil_marry(c, s));
+
+	// ridiculous hop count
+	TEST_NNG_PASS(nng_msg_alloc(&m, 0));
+	TEST_NNG_PASS(nng_msg_append_u32(m, 0x1000));
+	TEST_NNG_PASS(nng_sendmsg(c, m, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s, &m, 0), NNG_ETIMEDOUT);
+
+	TEST_NNG_PASS(nng_close(c));
+	TEST_NNG_PASS(nng_close(s));
+}
+
 TEST_LIST = {
 	{ "pair1 monogamous cooked", test_mono_cooked },
 	{ "pair1 monogamous faithful", test_mono_faithful },
 	{ "pair1 monogamous back pressure", test_mono_back_pressure },
 	{ "pair1 monogamous raw exchange", test_mono_raw_exchange },
 	{ "pair1 monogamous raw header", test_mono_raw_header },
-	{ "pair1 polyamorous best effort", test_poly_best_effort },
-	{ "pair1 polyamorous cooked", test_poly_cooked },
-	{ "pair1 polyamorous late", test_poly_late },
-	{ "pair1 polyamorous default", test_poly_default },
-	{ "pair1 polyamorous raw", test_poly_raw },
-	{ "pair1 raw", test_raw },
-	{ "pair1 ttl", test_ttl },
-	{ "pair1 validate peer", test_validate_peer },
+	{ "pair1 raw", test_pair1_raw },
+	{ "pair1 ttl", test_pair1_ttl },
+	{ "pair1 validate peer", test_pair1_validate_peer },
+	{ "pair1 recv no header", test_pair1_recv_no_header },
+	{ "pair1 recv garbage", test_pair1_recv_garbage },
 
 	{ NULL, NULL },
 };