- update smtpctl.8 to reflect reality

- bring back 'smtpctl schedule' and 'smtpctl remove' to life

Things you should know:

The ramqueue data structure is not finished yet and lacks an envelope tree
for evpid lookups. I wanted to wait until I'm done but too many people are
affected by not being able to reschedule envelopes, this is a quick fix.

So right now there's an O(rrible) complexity as both commands will perform
a (possibly aborted) queue scan leading to O(n). I will make that O(log n)
soon.

Also, smtpctl remove no longer supports removing an entire message, I will
fix that very soon too.

8 files changed, 165 insertions, 21 deletions

diff --git a/usr.sbin/smtpd/control.c b/usr.sbin/smtpd/control.c
index 28800a90f64..52749cfc2a7 100644
--- a/usr.sbin/smtpd/control.c
+++ b/usr.sbin/smtpd/control.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: control.c,v 1.58 2011/05/01 12:57:11 eric Exp $	*/
+/*	$OpenBSD: control.c,v 1.59 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -121,6 +121,7 @@ control(void)
 	struct event		 ev_sigint;
 	struct event		 ev_sigterm;
 	struct peer		 peers [] = {
+		{ PROC_RUNNER,	 imsg_dispatch },
 		{ PROC_QUEUE,	 imsg_dispatch },
 		{ PROC_SMTP,	 imsg_dispatch },
 		{ PROC_MFA,	 imsg_dispatch },
@@ -472,6 +473,7 @@ control_dispatch_ext(int fd, short event, void *arg)
 			    IMSG_QUEUE_RESUME_OUTGOING, 0, 0, -1, NULL, 0);
 			imsg_compose_event(&c->iev, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 			break;
+
 		case IMSG_SMTP_RESUME:
 			if (euid)
 				goto badcred;
@@ -486,6 +488,36 @@ control_dispatch_ext(int fd, short event, void *arg)
 			    0, 0, -1, NULL, 0);
 			imsg_compose_event(&c->iev, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
 			break;
+
+		case IMSG_RUNNER_SCHEDULE: {
+			u_int64_t ullval;
+
+			if (euid)
+				goto badcred;
+
+			ullval = *(u_int64_t *)imsg.data;
+
+			imsg_compose_event(env->sc_ievs[PROC_RUNNER], IMSG_RUNNER_SCHEDULE,
+			    0, 0, -1, &ullval, sizeof(ullval));
+
+			imsg_compose_event(&c->iev, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
+			break;
+		}
+
+		case IMSG_RUNNER_REMOVE: {
+			u_int64_t ullval;
+
+			if (euid)
+				goto badcred;
+
+			ullval = *(u_int64_t *)imsg.data;
+
+			imsg_compose_event(env->sc_ievs[PROC_RUNNER], IMSG_RUNNER_REMOVE,
+			    0, 0, -1, &ullval, sizeof(ullval));
+
+			imsg_compose_event(&c->iev, IMSG_CTL_OK, 0, 0, -1, NULL, 0);
+			break;
+		}
 		default:
 			log_debug("control_dispatch_ext: "
 			    "error handling imsg %d", imsg.hdr.type);
diff --git a/usr.sbin/smtpd/parser.c b/usr.sbin/smtpd/parser.c
index bc9e0fac9b1..dc8032115ba 100644
--- a/usr.sbin/smtpd/parser.c
+++ b/usr.sbin/smtpd/parser.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: parser.c,v 1.19 2011/04/13 20:53:18 gilles Exp $	*/
+/*	$OpenBSD: parser.c,v 1.20 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -48,22 +48,36 @@ struct token {
 };
 
 static const struct token t_main[];
+static const struct token t_schedule[];
 static const struct token t_show[];
 static const struct token t_pause[];
+static const struct token t_remove[];
 static const struct token t_resume[];
 static const struct token t_log[];
 
 static const struct token t_main[] = {
+	{KEYWORD,	"schedule",    	NONE,		t_schedule},
 	{KEYWORD,	"show",		NONE,		t_show},
 	{KEYWORD,	"monitor",	MONITOR,	NULL},
 	{KEYWORD,	"pause",	NONE,      	t_pause},
 /*	{KEYWORD,	"reload",	RELOAD,		NULL},*/
+	{KEYWORD,	"remove",	NONE,      	t_remove},
 	{KEYWORD,	"resume",	NONE,      	t_resume},
 	{KEYWORD,	"stop",		SHUTDOWN,      	NULL},
 	{KEYWORD,	"log",    	NONE,      	t_log},
 	{ENDTOKEN,	"",		NONE,		NULL}
 };
 
+static const struct token t_remove[] = {
+	{VARIABLE,	"evpid",	REMOVE,		NULL},
+	{ENDTOKEN,	"",		NONE,		NULL}
+};
+
+static const struct token t_schedule[] = {
+	{VARIABLE,	"msgid/evpid",	SCHEDULE,	NULL},
+	{ENDTOKEN,	"",		NONE,		NULL}
+};
+
 static const struct token t_show[] = {
 	{KEYWORD,	"queue",	SHOW_QUEUE,	NULL},
 	{KEYWORD,	"runqueue",	SHOW_RUNQUEUE,	NULL},
diff --git a/usr.sbin/smtpd/parser.h b/usr.sbin/smtpd/parser.h
index 06229ccb929..694fb0096ed 100644
--- a/usr.sbin/smtpd/parser.h
+++ b/usr.sbin/smtpd/parser.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: parser.h,v 1.16 2011/04/13 20:53:18 gilles Exp $	*/
+/*	$OpenBSD: parser.h,v 1.17 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -23,12 +23,14 @@ enum actions {
 	MONITOR,
 	LOG_VERBOSE,
 	LOG_BRIEF,
+	SCHEDULE,
 	SHOW_QUEUE,
 	SHOW_RUNQUEUE,
 	SHOW_STATS,
 	PAUSE_MDA,
 	PAUSE_MTA,
 	PAUSE_SMTP,
+	REMOVE,
 	RESUME_MDA,
 	RESUME_MTA,
 	RESUME_SMTP,
diff --git a/usr.sbin/smtpd/ramqueue.c b/usr.sbin/smtpd/ramqueue.c
index 40765735053..3df6c66b074 100644
--- a/usr.sbin/smtpd/ramqueue.c
+++ b/usr.sbin/smtpd/ramqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: ramqueue.c,v 1.9 2011/07/19 13:07:07 eric Exp $	*/
+/*	$OpenBSD: ramqueue.c,v 1.10 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -353,4 +353,43 @@ ramqueue_host_cmp(struct ramqueue_host *h1, struct ramqueue_host *h2)
 	return strcmp(h1->hostname, h2->hostname);
 }
 
+void
+ramqueue_reschedule(struct ramqueue *rqueue, u_int64_t id)
+{
+	struct ramqueue_envelope *rq_evp;
+	time_t tm;
+
+	tm = time(NULL);
+	TAILQ_FOREACH(rq_evp, &rqueue->queue, queue_entry) {
+		if (id != rq_evp->evpid &&
+		    (id <= 0xffffffffLL &&
+			evpid_to_msgid(rq_evp->evpid) != id))
+			continue;
+
+		TAILQ_REMOVE(&rqueue->queue, rq_evp, queue_entry);
+		rq_evp->sched = 0;
+		TAILQ_INSERT_HEAD(&rqueue->queue, rq_evp, queue_entry);
+
+		/* we were scheduling one envelope and found it,
+		 * no need to go through the entire queue
+		 */
+		if (id > 0xffffffffLL)
+			break;
+	}
+}
+
+struct ramqueue_envelope *
+ramqueue_envelope_by_id(struct ramqueue *rqueue, u_int64_t id)
+{
+	struct ramqueue_envelope *rq_evp;
+
+	TAILQ_FOREACH(rq_evp, &rqueue->queue, queue_entry) {
+		if (rq_evp->evpid == id)
+			return rq_evp;
+	}
+
+	return NULL;
+}
+
 RB_GENERATE(hosttree, ramqueue_host, host_entry, ramqueue_host_cmp);
+
diff --git a/usr.sbin/smtpd/runner.c b/usr.sbin/smtpd/runner.c
index cb9128a633c..043292218c1 100644
--- a/usr.sbin/smtpd/runner.c
+++ b/usr.sbin/smtpd/runner.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: runner.c,v 1.107 2011/05/16 21:05:52 gilles Exp $	*/
+/*	$OpenBSD: runner.c,v 1.108 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -155,6 +155,32 @@ runner_imsg(struct imsgev *iev, struct imsg *imsg)
 	case IMSG_CTL_VERBOSE:
 		log_verbose(*(int *)imsg->data);
 		return;
+
+	case IMSG_RUNNER_SCHEDULE:
+		ramqueue_reschedule(&env->sc_rqueue,
+		    *(u_int64_t *)imsg->data);
+		runner_reset_events();		
+		return;
+
+	case IMSG_RUNNER_REMOVE: {
+		u_int64_t ullval = *(u_int64_t *)imsg->data;
+		struct envelope	 envelope;
+		struct ramqueue_envelope *rq_evp;
+		
+		if (! queue_envelope_load(Q_QUEUE, ullval, &envelope))
+			return;
+		
+		rq_evp = ramqueue_envelope_by_id(&env->sc_rqueue, ullval);
+		if (rq_evp == NULL)
+			return;
+
+		ramqueue_remove(&env->sc_rqueue, rq_evp);
+
+		queue_envelope_delete(Q_QUEUE, &envelope);
+
+		runner_reset_events();
+		return;
+	}
 	}
 
 	fatalx("runner_imsg: unexpected imsg");
@@ -217,6 +243,7 @@ runner(void)
 	struct event	 ev_sigterm;
 
 	struct peer peers[] = {
+		{ PROC_CONTROL,	imsg_dispatch },
 		{ PROC_QUEUE,	imsg_dispatch }
 	};
 
diff --git a/usr.sbin/smtpd/smtpctl.8 b/usr.sbin/smtpd/smtpctl.8
index 294407e7a7e..dff08c88d0c 100644
--- a/usr.sbin/smtpd/smtpctl.8
+++ b/usr.sbin/smtpd/smtpctl.8
@@ -1,4 +1,4 @@
-.\" $OpenBSD: smtpctl.8,v 1.18 2010/10/18 13:28:25 sthen Exp $
+.\" $OpenBSD: smtpctl.8,v 1.19 2011/07/21 23:29:24 gilles Exp $
 .\"
 .\" Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
 .\"
@@ -14,7 +14,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 18 2010 $
+.Dd $Mdocdate: July 21 2011 $
 .Dt SMTPCTL 8
 .Os
 .Sh NAME
@@ -47,16 +47,16 @@ Temporarily stop deliveries to local users.
 .It Cm pause outgoing
 Temporarily stop relaying and deliveries to
 remote users.
-.It Cm remove Ar message-uid | message-id
-Removes a single message, or all messages with the same message ID.
+.It Cm remove Ar envelope-id
+Removes a single envelope from queue.
 .It Cm resume incoming
 Resume accepting incoming sessions.
 .It Cm resume local
 Resume deliveries to local users.
 .It Cm resume outgoing
 Resume relaying and deliveries to remote users.
-.It Cm schedule Ar message-uid | message-id
-Marks a single message, or all messages with the same message ID,
+.It Cm schedule Ar envelope-id | message-id
+Marks a single envelope, or all envelopes with the same message ID,
 as ready for immediate delivery.
 .It Cm show queue
 Displays information concerning envelopes
@@ -74,24 +74,24 @@ When
 receives a message, it generates a
 .Ar message-id
 for the message, and one
-.Ar message-uid
+.Ar envelope-id
 per recipient.
 The
 .Ar message-id
-consists of a timestamp followed by a random identifier
-that is guaranteed to be unique on the host system.
+is a 32 bits random identifier that is guaranteed to be
+unique on the host system.
 The
-.Ar message-uid
-consists of the
+.Ar envelope-id
+is a 64 bits unique identifier that encodes the
 .Ar message-id
-followed by a random identifier that is also guaranteed
-to be unique between recipients of the same message.
+in the 32 upper bits and a random envelope identifier
+in the 32 lower bits.
 .Pp
 A command which specifies a
 .Ar message-id
 applies to all recipients of a message;
 a command which specifies a
-.Ar message-uid
+.Ar envelope-uid
 applies to a specific recipient of a message.
 .Sh FILES
 .Bl -tag -width "/var/run/smtpd.sockXX" -compact
diff --git a/usr.sbin/smtpd/smtpctl.c b/usr.sbin/smtpd/smtpctl.c
index e785d6434b0..8c13c3b52cc 100644
--- a/usr.sbin/smtpd/smtpctl.c
+++ b/usr.sbin/smtpd/smtpctl.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpctl.c,v 1.60 2011/05/01 12:57:11 eric Exp $	*/
+/*	$OpenBSD: smtpctl.c,v 1.61 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -27,6 +27,7 @@
 #include <sys/param.h>
 
 #include <err.h>
+#include <errno.h>
 #include <event.h>
 #include <imsg.h>
 #include <stdio.h>
@@ -128,6 +129,28 @@ connected:
 	case NONE:
 		usage();
 		/* not reached */
+
+	case SCHEDULE:
+	case REMOVE: {
+		u_int64_t ulval;
+		char *ep;
+
+		errno = 0;
+		ulval = strtoul(res->data, &ep, 16);
+		if (res->data[0] == '\0' || *ep != '\0')
+			errx(1, "invalid msgid/evpid");
+		if (errno == ERANGE && ulval == ULLONG_MAX)
+			errx(1, "invalid msgid/evpid");
+
+		if (res->action == SCHEDULE)
+			imsg_compose(ibuf, IMSG_RUNNER_SCHEDULE, 0, 0, -1, &ulval,
+			    sizeof(ulval));
+		if (res->action == REMOVE)
+			imsg_compose(ibuf, IMSG_RUNNER_REMOVE, 0, 0, -1, &ulval,
+			    sizeof(ulval));
+		break;
+	}
+
 	case SHUTDOWN:
 		imsg_compose(ibuf, IMSG_CTL_SHUTDOWN, 0, 0, -1, NULL, 0);
 		break;
@@ -185,6 +208,8 @@ connected:
 				break;
 			switch(res->action) {
 			/* case RELOAD: */
+			case REMOVE:
+			case SCHEDULE:
 			case SHUTDOWN:
 			case PAUSE_MDA:
 			case PAUSE_MTA:
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index 1be8b5c8c8b..06acadc9143 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.228 2011/07/20 10:22:54 eric Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.229 2011/07/21 23:29:24 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -158,6 +158,9 @@ enum imsg_type {
 	IMSG_QUEUE_SCHEDULE,
 	IMSG_QUEUE_REMOVE,
 
+	IMSG_RUNNER_REMOVE,
+	IMSG_RUNNER_SCHEDULE,
+
 	IMSG_BATCH_CREATE,
 	IMSG_BATCH_APPEND,
 	IMSG_BATCH_CLOSE,
@@ -1089,6 +1092,8 @@ int ramqueue_batch_is_empty(struct ramqueue_batch *);
 int ramqueue_host_is_empty(struct ramqueue_host *);
 void ramqueue_remove_batch(struct ramqueue_host *, struct ramqueue_batch *);
 void ramqueue_remove_host(struct ramqueue *, struct ramqueue_host *);
+void ramqueue_reschedule(struct ramqueue *, u_int64_t);
+struct ramqueue_envelope *ramqueue_envelope_by_id(struct ramqueue *, u_int64_t);
 struct ramqueue_envelope *ramqueue_first_envelope(struct ramqueue *);
 struct ramqueue_envelope *ramqueue_next_envelope(struct ramqueue *);
 struct ramqueue_envelope *ramqueue_batch_first_envelope(struct ramqueue_batch *);