Implement smtpctl discover <evpid|msgid>.

discover subcommand schedules envelopes manually moved to the queue.
It triggers a queue walk searching for envelopes with the given id,
schedules them and informs the user number of envelopes scheduled.
Admins no longer would need to restart the daemon to discover
manually moved messages.

Ok gilles@

11 files changed, 350 insertions, 14 deletions

diff --git a/usr.sbin/smtpd/control.c b/usr.sbin/smtpd/control.c
index 362a8f9787d..2a0dec52f8c 100644
--- a/usr.sbin/smtpd/control.c
+++ b/usr.sbin/smtpd/control.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: control.c,v 1.106 2015/10/13 10:59:04 gilles Exp $	*/
+/*	$OpenBSD: control.c,v 1.107 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2012 Gilles Chehade <gilles@poolp.org>
@@ -116,6 +116,8 @@ control_imsg(struct mproc *p, struct imsg *imsg)
 	if (p->proc == PROC_QUEUE) {
 		switch (imsg->hdr.type) {
 		case IMSG_CTL_LIST_ENVELOPES:
+		case IMSG_CTL_DISCOVER_EVPID:
+		case IMSG_CTL_DISCOVER_MSGID:
 			c = tree_get(&ctl_conns, imsg->hdr.peerid);
 			if (c == NULL)
 				return;
@@ -471,6 +473,8 @@ control_dispatch_ext(struct mproc *p, struct imsg *imsg)
 	char			*key;
 	struct stat_value	 val;
 	size_t			 len;
+	uint64_t		 evpid;
+	uint32_t		 msgid;
 
 	c = p->data;
 
@@ -793,6 +797,32 @@ control_dispatch_ext(struct mproc *p, struct imsg *imsg)
 		m_compose(p, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 		return;
 
+	case IMSG_CTL_DISCOVER_EVPID:
+		if (c->euid)
+			goto badcred;
+
+		if (imsg->hdr.len - IMSG_HEADER_SIZE != sizeof evpid)
+			goto invalid;
+
+		memmove(&evpid, imsg->data, sizeof evpid);
+		m_create(p_queue, imsg->hdr.type, c->id, 0, -1);
+		m_add_evpid(p_queue, evpid);
+		m_close(p_queue);
+		return;
+
+	case IMSG_CTL_DISCOVER_MSGID:
+		if (c->euid)
+			goto badcred;
+
+		if (imsg->hdr.len - IMSG_HEADER_SIZE != sizeof msgid)
+			goto invalid;
+
+		memmove(&msgid, imsg->data, sizeof msgid);
+		m_create(p_queue, imsg->hdr.type, c->id, 0, -1);
+		m_add_msgid(p_queue, msgid);
+		m_close(p_queue);
+		return;
+
 	default:
 		log_debug("debug: control_dispatch_ext: "
 		    "error handling %s imsg",
diff --git a/usr.sbin/smtpd/queue.c b/usr.sbin/smtpd/queue.c
index eff14eb18b3..dbe12312660 100644
--- a/usr.sbin/smtpd/queue.c
+++ b/usr.sbin/smtpd/queue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue.c,v 1.169 2015/10/16 13:37:44 millert Exp $	*/
+/*	$OpenBSD: queue.c,v 1.170 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -47,6 +47,7 @@ static void queue_bounce(struct envelope *, struct delivery_bounce *);
 static void queue_shutdown(void);
 static void queue_sig_handler(int, short, void *);
 static void queue_log(const struct envelope *, const char *, const char *);
+static void queue_msgid_walk(int, short, void *);
 
 static size_t	flow_agent_hiwat = 10 * 1024 * 1024;
 static size_t	flow_agent_lowat =   1 * 1024 * 1024;
@@ -62,6 +63,8 @@ static void
 queue_imsg(struct mproc *p, struct imsg *imsg)
 {
 	struct delivery_bounce	 bounce;
+	struct msg_walkinfo	*wi;
+	struct timeval		 tv;
 	struct bounce_req_msg	*req_bounce;
 	struct envelope		 evp;
 	struct msg		 m;
@@ -69,6 +72,7 @@ queue_imsg(struct mproc *p, struct imsg *imsg)
 	uint64_t		 reqid, evpid, holdq;
 	uint32_t		 msgid;
 	time_t			 nexttry;
+	size_t			 n_evp;
 	int			 fd, mta_ext, ret, v, flags, code;
 
 	memset(&bounce, 0, sizeof(struct delivery_bounce));
@@ -500,6 +504,48 @@ queue_imsg(struct mproc *p, struct imsg *imsg)
 			m_end(&m);
 			profiling = v;
 			return;
+
+		case IMSG_CTL_DISCOVER_EVPID:
+			m_msg(&m, imsg);
+			m_get_evpid(&m, &evpid);
+			m_end(&m);
+			if (queue_envelope_load(evpid, &evp) == 0) {
+				log_warnx("queue: discover: failed to load "
+				    "envelope %016" PRIx64, evpid);
+				n_evp = 0;
+				m_compose(p_control, imsg->hdr.type,
+				    imsg->hdr.peerid, 0, -1,
+				    &n_evp, sizeof n_evp);
+				return;
+			}
+
+			m_create(p_scheduler, IMSG_QUEUE_DISCOVER_EVPID,
+			    0, 0, -1);
+			m_add_envelope(p_scheduler, &evp);
+			m_close(p_scheduler);
+
+			m_create(p_scheduler, IMSG_QUEUE_DISCOVER_MSGID,
+			    0, 0, -1);
+			m_add_msgid(p_scheduler, evpid_to_msgid(evpid));
+			m_close(p_scheduler);
+			n_evp = 1;
+			m_compose(p_control, imsg->hdr.type, imsg->hdr.peerid,
+			    0, -1, &n_evp, sizeof n_evp);
+			return;
+
+		case IMSG_CTL_DISCOVER_MSGID:
+			m_msg(&m, imsg);
+			m_get_msgid(&m, &msgid);
+			m_end(&m);
+			/* handle concurrent walk requests */
+			wi = xcalloc(1, sizeof *wi, "queu_imsg");
+			wi->msgid = msgid;
+			wi->peerid = imsg->hdr.peerid;
+			evtimer_set(&wi->ev, queue_msgid_walk, wi);
+			tv.tv_sec = 0;
+			tv.tv_usec = 10;
+			evtimer_add(&wi->ev, &tv);
+			return;
 		}
 	}
 
@@ -507,6 +553,43 @@ queue_imsg(struct mproc *p, struct imsg *imsg)
 }
 
 static void
+queue_msgid_walk(int fd, short event, void *arg)
+{
+	struct envelope		 evp;
+	struct timeval		 tv;
+	struct msg_walkinfo	*wi = arg;
+	int			 r;
+
+	r = queue_message_walk(&evp, wi->msgid, &wi->done, &wi->data);
+	if (r == -1) {
+		if (wi->n_evp) {
+			m_create(p_scheduler, IMSG_QUEUE_DISCOVER_MSGID,
+			    0, 0, -1);
+			m_add_msgid(p_scheduler, wi->msgid);
+			m_close(p_scheduler);
+		}
+
+		m_compose(p_control, IMSG_CTL_DISCOVER_MSGID, wi->peerid, 0, -1,
+		    &wi->n_evp, sizeof wi->n_evp);
+		evtimer_del(&wi->ev);
+		free(wi);
+		return;
+	}
+
+	if (r) {
+		m_create(p_scheduler, IMSG_QUEUE_DISCOVER_EVPID, 0, 0, -1);
+		m_add_envelope(p_scheduler, &evp);
+		m_close(p_scheduler);
+		wi->n_evp += 1;
+	}
+
+	tv.tv_sec = 0;
+	tv.tv_usec = 10;
+	evtimer_set(&wi->ev, queue_msgid_walk, wi);
+	evtimer_add(&wi->ev, &tv);
+}
+
+static void
 queue_bounce(struct envelope *e, struct delivery_bounce *d)
 {
 	struct envelope	b;
diff --git a/usr.sbin/smtpd/queue_backend.c b/usr.sbin/smtpd/queue_backend.c
index ced1c81472d..17b77c48805 100644
--- a/usr.sbin/smtpd/queue_backend.c
+++ b/usr.sbin/smtpd/queue_backend.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue_backend.c,v 1.56 2015/10/09 14:37:38 gilles Exp $	*/
+/*	$OpenBSD: queue_backend.c,v 1.57 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@poolp.org>
@@ -70,6 +70,8 @@ static int (*handler_envelope_delete)(uint64_t);
 static int (*handler_envelope_update)(uint64_t, const char *, size_t);
 static int (*handler_envelope_load)(uint64_t, char *, size_t);
 static int (*handler_envelope_walk)(uint64_t *, char *, size_t);
+static int (*handler_message_walk)(uint64_t *, char *, size_t,
+    uint32_t, int *, void **);
 
 #ifdef QUEUE_PROFILING
 
@@ -619,6 +621,46 @@ queue_envelope_update(struct envelope *ep)
 }
 
 int
+queue_message_walk(struct envelope *ep, uint32_t msgid, int *done, void **data)
+{
+	char		 evpbuf[sizeof(struct envelope)];
+	uint64_t	 evpid;
+	int		 r;
+	const char	*e;
+
+	profile_enter("queue_message_walk");
+	r = handler_message_walk(&evpid, evpbuf, sizeof evpbuf,
+	    msgid, done, data);
+	profile_leave();
+
+	log_trace(TRACE_QUEUE,
+	    "queue-backend: queue_message_walk() -> %d (%016"PRIx64")",
+	    r, evpid);
+
+	if (r == -1)
+		return (r);
+
+	if (r && queue_envelope_load_buffer(ep, evpbuf, (size_t)r)) {
+		if ((e = envelope_validate(ep)) == NULL) {
+			ep->id = evpid;
+			/*
+			 * do not cache the envelope here, while discovering
+			 * envelopes one could re-run discover on already
+			 * scheduled envelopes which leads to triggering of 
+			 * strict checks in caching. Envelopes could anyway
+			 * be loaded from backend if it isn't cached.
+			 */
+			return (1);
+		}
+		log_debug("debug: invalid envelope %016" PRIx64 ": %s",
+		    ep->id, e);
+		(void)queue_message_corrupt(evpid_to_msgid(evpid));
+	}
+
+	return (-1);
+}
+
+int
 queue_envelope_walk(struct envelope *ep)
 {
 	const char	*e;
@@ -766,3 +808,10 @@ queue_api_on_envelope_walk(int(*cb)(uint64_t *, char *, size_t))
 {
 	handler_envelope_walk = cb;
 }
+
+void
+queue_api_on_message_walk(int(*cb)(uint64_t *, char *, size_t,
+    uint32_t, int *, void **))
+{
+	handler_message_walk = cb;
+}
diff --git a/usr.sbin/smtpd/queue_fs.c b/usr.sbin/smtpd/queue_fs.c
index 690c7f5d014..4063b477843 100644
--- a/usr.sbin/smtpd/queue_fs.c
+++ b/usr.sbin/smtpd/queue_fs.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue_fs.c,v 1.9 2015/10/12 22:29:49 gilles Exp $	*/
+/*	$OpenBSD: queue_fs.c,v 1.10 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@poolp.org>
@@ -24,6 +24,7 @@
 #include <sys/stat.h>
 
 #include <ctype.h>
+#include <dirent.h>
 #include <err.h>
 #include <errno.h>
 #include <event.h>
@@ -345,6 +346,69 @@ queue_fs_envelope_delete(uint64_t evpid)
 }
 
 static int
+queue_fs_message_walk(uint64_t *evpid, char *buf, size_t len,
+    uint32_t msgid, int *done, void **data)
+{
+	struct dirent	*dp;
+	DIR		*dir = *data;
+	char		 path[PATH_MAX];
+	char		 msgid_str[9];
+	char		*tmp;
+	int		 r, *n;
+
+	if (*done)
+		return (-1);
+
+	if (! bsnprintf(path, sizeof path, "%s/%02x/%08x",
+	    PATH_QUEUE, (msgid  & 0xff000000) >> 24, msgid))
+		fatalx("queue_fs_message_walk: path does not fit buffer");
+
+	if (dir == NULL) {
+		if ((dir = opendir(path)) == NULL) {
+			log_warn("warn: queue_fs: opendir: %s", path);
+			*done = 1;
+			return (-1);
+		}
+
+		*data = dir;
+	}
+
+	(void)snprintf(msgid_str, sizeof msgid_str, "%08" PRIx32, msgid);
+	while ((dp = readdir(dir)) != NULL) {
+		if (dp->d_type != DT_REG)
+			continue;
+
+		/* ignore files other than envelopes */
+		if (dp->d_namlen != 16 || strncmp(dp->d_name, msgid_str, 8))
+			continue;
+
+		tmp = NULL;
+		*evpid = strtoull(dp->d_name, &tmp, 16);
+		if (tmp && *tmp !=  '\0') {
+			log_debug("debug: fsqueue: bogus file %s", dp->d_name);
+			continue;
+		}
+
+		memset(buf, 0, len);
+		r = queue_fs_envelope_load(*evpid, buf, len);
+		if (r) {
+			n = tree_pop(&evpcount, msgid);
+			if (n == NULL)
+				n = REF;
+
+			n += 1;
+			tree_xset(&evpcount, msgid, n);
+		}
+
+		return (r);
+	}
+
+	(void)closedir(dir);
+	*done = 1;
+	return (-1);
+}
+
+static int
 queue_fs_envelope_walk(uint64_t *evpid, char *buf, size_t len)
 {
 	static int	 done = 0;
@@ -653,6 +717,7 @@ queue_fs_init(struct passwd *pw, int server, const char *conf)
 	queue_api_on_envelope_update(queue_fs_envelope_update);
 	queue_api_on_envelope_load(queue_fs_envelope_load);
 	queue_api_on_envelope_walk(queue_fs_envelope_walk);
+	queue_api_on_message_walk(queue_fs_message_walk);
 
 	return (ret);
 }
diff --git a/usr.sbin/smtpd/scheduler.c b/usr.sbin/smtpd/scheduler.c
index 171b99d3de1..94c0a608610 100644
--- a/usr.sbin/smtpd/scheduler.c
+++ b/usr.sbin/smtpd/scheduler.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: scheduler.c,v 1.49 2015/10/13 08:09:25 gilles Exp $	*/
+/*	$OpenBSD: scheduler.c,v 1.50 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -100,6 +100,41 @@ scheduler_imsg(struct mproc *p, struct imsg *imsg)
 		scheduler_reset_events();
 		return;
 
+	case IMSG_QUEUE_DISCOVER_EVPID:
+		m_msg(&m, imsg);
+		m_get_envelope(&m, &evp);
+		m_end(&m);
+		r = backend->query(evp.id);
+		if (r) {
+			log_debug("debug: scheduler: evp:%016" PRIx64
+			    " already scheduled", evp.id);
+			return;
+		}
+		log_trace(TRACE_SCHEDULER,
+		    "scheduler: discovering evp:%016" PRIx64, evp.id);
+		scheduler_info(&si, &evp);
+		stat_increment("scheduler.envelope.incoming", 1);
+		backend->insert(&si);
+		return;
+
+	case IMSG_QUEUE_DISCOVER_MSGID:
+		m_msg(&m, imsg);
+		m_get_msgid(&m, &msgid);
+		m_end(&m);
+		r = backend->query(msgid);
+		if (r) {
+			log_debug("debug: scheduler: msgid:%08" PRIx32
+			    " already scheduled", msgid);
+			return;
+		}
+		log_trace(TRACE_SCHEDULER,
+		    "scheduler: committing msg:%08" PRIx32, msgid);
+		n = backend->commit(msgid);
+		stat_decrement("scheduler.envelope.incoming", n);
+		stat_increment("scheduler.envelope", n);
+		scheduler_reset_events();
+		return;
+
 	case IMSG_QUEUE_MESSAGE_ROLLBACK:
 		m_msg(&m, imsg);
 		m_get_msgid(&m, &msgid);
diff --git a/usr.sbin/smtpd/scheduler_ramqueue.c b/usr.sbin/smtpd/scheduler_ramqueue.c
index 9ff16f16afd..d8d7c05ed8f 100644
--- a/usr.sbin/smtpd/scheduler_ramqueue.c
+++ b/usr.sbin/smtpd/scheduler_ramqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: scheduler_ramqueue.c,v 1.41 2015/01/20 17:37:54 deraadt Exp $	*/
+/*	$OpenBSD: scheduler_ramqueue.c,v 1.42 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2012 Gilles Chehade <gilles@poolp.org>
@@ -114,6 +114,7 @@ static int scheduler_ram_schedule(uint64_t);
 static int scheduler_ram_remove(uint64_t);
 static int scheduler_ram_suspend(uint64_t);
 static int scheduler_ram_resume(uint64_t);
+static int scheduler_ram_query(uint64_t);
 
 static void sorted_insert(struct rq_queue *, struct rq_envelope *);
 
@@ -149,6 +150,7 @@ struct scheduler_backend scheduler_backend_ramqueue = {
 	scheduler_ram_remove,
 	scheduler_ram_suspend,
 	scheduler_ram_resume,
+	scheduler_ram_query,
 };
 
 static struct rq_queue	ramqueue;
@@ -801,6 +803,22 @@ scheduler_ram_resume(uint64_t evpid)
 	}
 }
 
+static int
+scheduler_ram_query(uint64_t evpid)
+{
+	uint32_t msgid;
+
+	if (evpid > 0xffffffff)
+		msgid = evpid_to_msgid(evpid);
+	else
+		msgid = evpid;
+
+	if (tree_get(&ramqueue.messages, msgid) == NULL)
+		return (0);
+
+	return (1);
+}
+
 static void
 sorted_insert(struct rq_queue *rq, struct rq_envelope *evp)
 {
diff --git a/usr.sbin/smtpd/smtpctl.8 b/usr.sbin/smtpd/smtpctl.8
index add7b3159c8..a7ca06300e6 100644
--- a/usr.sbin/smtpd/smtpctl.8
+++ b/usr.sbin/smtpd/smtpctl.8
@@ -1,4 +1,4 @@
-.\" $OpenBSD: smtpctl.8,v 1.52 2015/10/15 08:18:23 sunil Exp $
+.\" $OpenBSD: smtpctl.8,v 1.53 2015/10/29 10:25:36 sunil Exp $
 .\"
 .\" Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
 .\" Copyright (c) 2012 Gilles Chehade <gilles@poolp.org>
@@ -15,7 +15,7 @@
 .\" ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 .\" OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 .\"
-.Dd $Mdocdate: October 15 2015 $
+.Dd $Mdocdate: October 29 2015 $
 .Dt SMTPCTL 8
 .Os
 .Sh NAME
@@ -45,6 +45,9 @@ and is simply a shortcut for
 .Pp
 The following commands are available:
 .Bl -tag -width Ds
+.It Cm discover Ar envelope-id | message-id
+Schedule a single envelope, or all envelopes with the same message ID
+that were manually moved to the queue.
 .It Cm encrypt Op Ar string
 Encrypt the password
 .Ar string
diff --git a/usr.sbin/smtpd/smtpctl.c b/usr.sbin/smtpd/smtpctl.c
index 964b5f5a4ef..d190a8e9096 100644
--- a/usr.sbin/smtpd/smtpctl.c
+++ b/usr.sbin/smtpd/smtpctl.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpctl.c,v 1.134 2015/10/28 07:24:28 sunil Exp $	*/
+/*	$OpenBSD: smtpctl.c,v 1.135 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2013 Eric Faurot <eric@openbsd.org>
@@ -910,6 +910,38 @@ do_show_mta_block(int argc, struct parameter *argv)
 	return (0);
 }
 
+static int
+do_discover(int argc, struct parameter *argv)
+{
+	uint64_t evpid;
+	uint32_t msgid;
+	size_t	 n_evp;
+
+	if (ibuf == NULL && !srv_connect())
+		errx(1, "smtpd doesn't seem to be running");
+
+	if (argv[0].type == P_EVPID) {
+		evpid = argv[0].u.u_evpid;
+		srv_send(IMSG_CTL_DISCOVER_EVPID, &evpid, sizeof evpid);
+		srv_recv(IMSG_CTL_DISCOVER_EVPID);
+	} else {
+		msgid = argv[0].u.u_msgid;
+		srv_send(IMSG_CTL_DISCOVER_MSGID, &msgid, sizeof msgid);
+		srv_recv(IMSG_CTL_DISCOVER_MSGID);
+	}
+
+	if (rlen == 0) {
+		srv_end();
+		return (0);
+	} else {
+		srv_read(&n_evp, sizeof n_evp);
+		srv_end();
+	}
+	
+	printf("%zu envelope%s discovered\n", n_evp, (n_evp != 1) ? "s" : "");
+	return (0);
+}
+
 int
 main(int argc, char **argv)
 {
@@ -941,6 +973,8 @@ main(int argc, char **argv)
 	if (setresgid(gid, gid, gid) == -1)
 		err(1, "setresgid");
 
+	cmd_install("discover <evpid>",		do_discover);
+	cmd_install("discover <msgid>",		do_discover);
 	cmd_install("encrypt",			do_encrypt);
 	cmd_install("encrypt <str>",		do_encrypt);
 	cmd_install("pause mta from <addr> for <str>", do_block_mta);
diff --git a/usr.sbin/smtpd/smtpd-api.h b/usr.sbin/smtpd/smtpd-api.h
index 6cb1a10e796..98b15b29b75 100644
--- a/usr.sbin/smtpd/smtpd-api.h
+++ b/usr.sbin/smtpd/smtpd-api.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd-api.h,v 1.22 2015/10/14 21:02:11 gilles Exp $	*/
+/*	$OpenBSD: smtpd-api.h,v 1.23 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2013 Eric Faurot <eric@openbsd.org>
@@ -356,6 +356,8 @@ void queue_api_on_envelope_delete(int(*)(uint64_t));
 void queue_api_on_envelope_update(int(*)(uint64_t, const char *, size_t));
 void queue_api_on_envelope_load(int(*)(uint64_t, char *, size_t));
 void queue_api_on_envelope_walk(int(*)(uint64_t *, char *, size_t));
+void queue_api_on_message_walk(int(*)(uint64_t *, char *, size_t,
+    uint32_t, int *, void **));
 void queue_api_no_chroot(void);
 void queue_api_set_chroot(const char *);
 void queue_api_set_user(const char *);
diff --git a/usr.sbin/smtpd/smtpd.c b/usr.sbin/smtpd/smtpd.c
index b46c9f7ad8b..93d6b3eb49e 100644
--- a/usr.sbin/smtpd/smtpd.c
+++ b/usr.sbin/smtpd/smtpd.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.c,v 1.252 2015/10/26 09:56:01 jung Exp $	*/
+/*	$OpenBSD: smtpd.c,v 1.253 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -1423,6 +1423,8 @@ imsg_to_str(int type)
 	CASE(IMSG_CTL_TRACE_ENABLE);
 	CASE(IMSG_CTL_UPDATE_TABLE);
 	CASE(IMSG_CTL_VERBOSE);
+	CASE(IMSG_CTL_DISCOVER_EVPID);
+	CASE(IMSG_CTL_DISCOVER_MSGID);
 
 	CASE(IMSG_CTL_SMTP_SESSION);
 
@@ -1443,6 +1445,8 @@ imsg_to_str(int type)
 	CASE(IMSG_QUEUE_DELIVERY_TEMPFAIL);
 	CASE(IMSG_QUEUE_DELIVERY_PERMFAIL);
 	CASE(IMSG_QUEUE_DELIVERY_LOOP);
+	CASE(IMSG_QUEUE_DISCOVER_EVPID);
+	CASE(IMSG_QUEUE_DISCOVER_MSGID);
 	CASE(IMSG_QUEUE_ENVELOPE_ACK);
 	CASE(IMSG_QUEUE_ENVELOPE_COMMIT);
 	CASE(IMSG_QUEUE_ENVELOPE_REMOVE);
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index 6a4b99acf74..42316a94846 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.482 2015/10/28 14:30:03 gilles Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.483 2015/10/29 10:25:36 sunil Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@poolp.org>
@@ -152,7 +152,7 @@ union lookup {
  * Bump IMSG_VERSION whenever a change is made to enum imsg_type.
  * This will ensure that we can never use a wrong version of smtpctl with smtpd.
  */
-#define	IMSG_VERSION		11
+#define	IMSG_VERSION		13
 
 enum imsg_type {
 	IMSG_NONE,
@@ -191,6 +191,8 @@ enum imsg_type {
 	IMSG_CTL_TRACE_ENABLE,
 	IMSG_CTL_UPDATE_TABLE,
 	IMSG_CTL_VERBOSE,
+	IMSG_CTL_DISCOVER_EVPID,
+	IMSG_CTL_DISCOVER_MSGID,
 
 	IMSG_CTL_SMTP_SESSION,
 
@@ -211,6 +213,8 @@ enum imsg_type {
 	IMSG_QUEUE_DELIVERY_TEMPFAIL,
 	IMSG_QUEUE_DELIVERY_PERMFAIL,
 	IMSG_QUEUE_DELIVERY_LOOP,
+	IMSG_QUEUE_DISCOVER_EVPID,
+	IMSG_QUEUE_DISCOVER_MSGID,
 	IMSG_QUEUE_ENVELOPE_ACK,
 	IMSG_QUEUE_ENVELOPE_COMMIT,
 	IMSG_QUEUE_ENVELOPE_REMOVE,
@@ -898,6 +902,7 @@ struct scheduler_backend {
 	int	(*remove)(uint64_t);
 	int	(*suspend)(uint64_t);
 	int	(*resume)(uint64_t);
+	int	(*query)(uint64_t);
 };
 
 enum stat_type {
@@ -1068,6 +1073,14 @@ struct ca_vrfy_resp_msg {
 	enum ca_resp_status	status;
 };
 
+struct msg_walkinfo {
+	struct event	 ev;
+	uint32_t	 msgid;
+	uint32_t	 peerid;
+	size_t		 n_evp;
+	void		*data;
+	int		 done;
+};
 
 /* aliases.c */
 int aliases_get(struct expand *, const char *);
@@ -1294,7 +1307,7 @@ int queue_envelope_delete(uint64_t);
 int queue_envelope_load(uint64_t, struct envelope *);
 int queue_envelope_update(struct envelope *);
 int queue_envelope_walk(struct envelope *);
-
+int queue_message_walk(struct envelope *, uint32_t, int *, void **);
 
 /* ruleset.c */
 struct rule *ruleset_match(const struct envelope *);