Qwalk, our API to linearly walk over the persistent queue, did not take the

queue_backend into account and assumed a filesystem with a specific layout.

This commit does plenty of things:

- make qwalk an abstraction in the queue_backend API, and impose queue
  drivers to implement qwalk_open(), qwalk() and qwalk_close();

- move previous qwalk_open(), qwalk() and qwalk_close() to the fsqueue
  driver since they were fsqueue specific ...

- make qwalk API work with msgid/evpid instead of pathnames since we're
  going to use the queue_backend API to load envelopes by evpid anyway;

- makes smtpd use *solely* the queue_backend API when manipulating the
  queue. pathnames were removed from smtpd.h and moved into the fsqueue
  which means we can now store a queue anywhere ... as long as we write
  the ten functions or so required for a queue driver ;-)

ok eric@, ok chl@

8 files changed, 380 insertions, 342 deletions

diff --git a/usr.sbin/smtpd/queue.c b/usr.sbin/smtpd/queue.c
index 4c175fa39b5..642143e0fa1 100644
--- a/usr.sbin/smtpd/queue.c
+++ b/usr.sbin/smtpd/queue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue.c,v 1.111 2011/11/14 19:23:41 chl Exp $	*/
+/*	$OpenBSD: queue.c,v 1.112 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -42,7 +42,7 @@ static void queue_imsg(struct imsgev *, struct imsg *);
 static void queue_pass_to_runner(struct imsgev *, struct imsg *);
 static void queue_shutdown(void);
 static void queue_sig_handler(int, short, void *);
-static void queue_purge(enum queue_kind, char *);
+static void queue_purge(enum queue_kind);
 
 static void
 queue_imsg(struct imsgev *iev, struct imsg *imsg)
@@ -302,8 +302,8 @@ queue(void)
 	config_pipes(peers, nitems(peers));
 	config_peers(peers, nitems(peers));
 
-	queue_purge(Q_INCOMING, PATH_INCOMING);
-	queue_purge(Q_ENQUEUE, PATH_ENQUEUE);
+	queue_purge(Q_INCOMING);
+	queue_purge(Q_ENQUEUE);
 
 	if (event_dispatch() <  0)
 		fatal("event_dispatch");
@@ -313,23 +313,17 @@ queue(void)
 }
 
 static void
-queue_purge(enum queue_kind qkind, char *queuepath)
+queue_purge(enum queue_kind qkind)
 {
-	char		 path[MAXPATHLEN];
 	struct qwalk	*q;
+	u_int32_t	 msgid;
+	u_int64_t	 evpid;
 
-	q = qwalk_new(queuepath);
-
-	while (qwalk(q, path)) {
-		u_int32_t msgid;
-
-		if ((msgid = filename_to_msgid(basename(path))) == 0) {
-			log_warnx("queue_purge: invalid evpid");
-			continue;
-		}
+	q = qwalk_new(qkind, 0);
+	while (qwalk(q, &evpid)) {
+		msgid = evpid_to_msgid(evpid);
 		queue_message_purge(qkind, msgid);
 	}
-
 	qwalk_close(q);
 }
 
diff --git a/usr.sbin/smtpd/queue_backend.c b/usr.sbin/smtpd/queue_backend.c
index 564baf47496..94c6a4b4b59 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.13 2011/10/23 13:03:04 gilles Exp $	*/
+/*	$OpenBSD: queue_backend.c,v 1.14 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -42,12 +42,19 @@ static int envelope_validate(struct envelope *);
 int	fsqueue_init(void);
 int	fsqueue_message(enum queue_kind, enum queue_op, u_int32_t *);
 int	fsqueue_envelope(enum queue_kind, enum queue_op , struct envelope *);
+void   *fsqueue_qwalk_new(enum queue_kind, u_int32_t);
+int	fsqueue_qwalk(void *, u_int64_t *);
+void	fsqueue_qwalk_close(void *);
+
 
 struct queue_backend queue_backends[] = {
 	{ QT_FS,
 	  fsqueue_init,
 	  fsqueue_message,
-	  fsqueue_envelope }
+	  fsqueue_envelope,
+	  fsqueue_qwalk_new,
+	  fsqueue_qwalk,
+	  fsqueue_qwalk_close }
 };
 
 struct queue_backend *
@@ -110,6 +117,7 @@ queue_message_fd_rw(enum queue_kind qkind, u_int32_t msgid)
 int
 queue_envelope_create(enum queue_kind qkind, struct envelope *ep)
 {
+	ep->id >>= 32;
 	return env->sc_queue->envelope(qkind, QOP_CREATE, ep);
 }
 
@@ -134,6 +142,57 @@ queue_envelope_update(enum queue_kind qkind, struct envelope *ep)
 	return env->sc_queue->envelope(qkind, QOP_UPDATE, ep);
 }
 
+void *
+qwalk_new(enum queue_kind kind, u_int32_t msgid)
+{
+	return env->sc_queue->qwalk_new(kind, msgid);
+}
+
+int
+qwalk(void *hdl, u_int64_t *evpid)
+{
+	return env->sc_queue->qwalk(hdl, evpid);
+}
+
+void
+qwalk_close(void *hdl)
+{
+	return env->sc_queue->qwalk_close(hdl);
+}
+
+u_int32_t
+queue_generate_msgid(void)
+{
+	u_int32_t msgid;
+
+again:
+	msgid = arc4random_uniform(0xffffffff);
+	if (msgid == 0)
+		goto again;
+
+	return msgid;
+}
+
+u_int64_t
+queue_generate_evpid(u_int32_t msgid)
+{
+	u_int32_t rnd;
+	u_int64_t evpid;
+
+again:
+	rnd = arc4random_uniform(0xffffffff);
+	if (rnd == 0)
+		goto again;
+
+	evpid = msgid;
+	evpid <<= 32;
+	evpid |= rnd;
+
+	return evpid;
+}
+
+
+/**/
 static int
 envelope_validate(struct envelope *ep)
 {
diff --git a/usr.sbin/smtpd/queue_fsqueue.c b/usr.sbin/smtpd/queue_fsqueue.c
index b1de423f397..4fa72f639c8 100644
--- a/usr.sbin/smtpd/queue_fsqueue.c
+++ b/usr.sbin/smtpd/queue_fsqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue_fsqueue.c,v 1.18 2011/11/14 11:53:10 eric Exp $	*/
+/*	$OpenBSD: queue_fsqueue.c,v 1.19 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -23,6 +23,7 @@
 #include <sys/socket.h>
 #include <sys/stat.h>
 
+#include <dirent.h>
 #include <err.h>
 #include <errno.h>
 #include <event.h>
@@ -39,6 +40,7 @@
 
 #include "smtpd.h"
 #include "log.h"
+#include "queue_fsqueue.h"
 
 static char		*fsqueue_getpath(enum queue_kind);
 
@@ -57,10 +59,13 @@ static int	fsqueue_message_corrupt(enum queue_kind, u_int32_t);
 int	fsqueue_init(void);
 int	fsqueue_message(enum queue_kind, enum queue_op, u_int32_t *);
 int	fsqueue_envelope(enum queue_kind, enum queue_op , struct envelope *);
-
 int	fsqueue_load_envelope_ascii(FILE *, struct envelope *);
 int	fsqueue_dump_envelope_ascii(FILE *, struct envelope *);
 
+void   *fsqueue_qwalk_new(enum queue_kind, u_int32_t);
+int	fsqueue_qwalk(void *, u_int64_t *);
+void	fsqueue_qwalk_close(void *);
+
 static char *
 fsqueue_getpath(enum queue_kind kind)
 {
@@ -77,9 +82,6 @@ fsqueue_getpath(enum queue_kind kind)
         case Q_PURGE:
                 return (PATH_PURGE);
 
-        case Q_OFFLINE:
-                return (PATH_OFFLINE);
-
         case Q_BOUNCE:
                 return (PATH_BOUNCE);
 
@@ -98,16 +100,12 @@ fsqueue_envelope_create(enum queue_kind qkind, struct envelope *ep)
 	char evpname[MAXPATHLEN];
 	FILE *fp;
 	int fd;
-	u_int32_t rnd;
 	u_int64_t evpid;
 
 	fp = NULL;
 
 again:
-	rnd = (u_int32_t)arc4random();
-	if (rnd == 0)
-		goto again;
-	evpid = ep->id | rnd;
+	evpid = queue_generate_evpid(ep->id);
 
 	if (! bsnprintf(evpname, sizeof(evpname), "%s/%08x%s/%016" PRIx64,
 		fsqueue_getpath(qkind),
@@ -279,7 +277,7 @@ fsqueue_message_create(enum queue_kind qkind, u_int32_t *msgid)
 	msgid_save = *msgid;
 
 again:
-	*msgid = (u_int32_t)arc4random();
+	*msgid = queue_generate_msgid();
 	if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%08x",
 		queuepath, *msgid))
 		fatalx("fsqueue_message_create: snprintf");
@@ -403,20 +401,44 @@ fsqueue_message_delete(enum queue_kind qkind, u_int32_t msgid)
 	char rootdir[MAXPATHLEN];
 	char evpdir[MAXPATHLEN];
 	char msgpath[MAXPATHLEN];
+	DIR *dirp;
+	struct dirent *dp;
 
-	if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%03x/%08x", PATH_QUEUE,
-		msgid & 0xfff, msgid))
-		fatal("queue_delete_message: snprintf");
+	if (qkind == Q_QUEUE) {
+		if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%03x/%08x", PATH_QUEUE,
+			msgid & 0xfff, msgid))
+			fatal("fsqueue_message_delete: snprintf");
+	}
+	else if (qkind == Q_PURGE) {
+		if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%08x", PATH_PURGE,
+			msgid))
+			fatal("fsqueue_message_delete: snprintf");
+	}
 
 	if (! bsnprintf(evpdir, sizeof(evpdir), "%s%s", rootdir,
 		PATH_ENVELOPES))
-		fatal("queue_delete_message: snprintf");
-	
+		fatal("fsqueue_message_delete: snprintf");
+
+	dirp = opendir(evpdir);
+	if (dirp) {
+		char envelope[MAXPATHLEN];
+
+		while ((dp = readdir(dirp)) != NULL) {
+			if (! bsnprintf(envelope, MAXPATHLEN, "%s/%s",
+				evpdir, dp->d_name))
+				fatal("fsqueue_message_delete: truncated evp");
+			unlink(envelope);
+		}
+		closedir(dirp);
+	}
+
 	if (! bsnprintf(msgpath, sizeof(msgpath), "%s/message", rootdir))
-		fatal("queue_delete_message: snprintf");
+		fatal("fsqueue_message_delete: snprintf");
 
-	if (unlink(msgpath) == -1)
-		fatal("queue_delete_message: unlink");
+	if (unlink(msgpath) == -1) {
+		if (errno != ENOENT)
+			fatal("fsqueue_message_delete: unlink");
+	}
 
 	if (rmdir(evpdir) == -1) {
 		/* It is ok to fail rmdir with ENOENT here
@@ -424,16 +446,18 @@ fsqueue_message_delete(enum queue_kind qkind, u_int32_t msgid)
 		 * last envelope, we remove the directory.
 		 */
 		if (errno != ENOENT)
-			fatal("queue_delete_message: rmdir");
+			fatal("fsqueue_message_delete: rmdir");
 	}
 
 	if (rmdir(rootdir) == -1)
-		fatal("#2 queue_delete_message: rmdir");
+		fatal("#2 fsqueue_message_delete: rmdir");
 
-	if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%03x", PATH_QUEUE, msgid & 0xffff))
-		fatal("queue_delete_message: snprintf");
 
-	rmdir(rootdir);
+	if (qkind == Q_QUEUE) {
+		if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%03x", PATH_QUEUE, msgid & 0xffff))
+			fatal("fsqueue_message_delete: snprintf");
+		rmdir(rootdir);
+	}
 
 	return 1;
 }
@@ -497,6 +521,12 @@ fsqueue_init(void)
 		if (ckdir(path, 0700, env->sc_pw->pw_uid, 0, 1) == 0)
 			ret = 0;
 	}
+
+	if (! bsnprintf(path, sizeof path, "%s/envelope.tmp", PATH_QUEUE))
+		errx(1, "path too long %s/envelope.tmp", PATH_QUEUE);
+
+	unlink(path);
+
 	return ret;
 }
 
@@ -554,3 +584,199 @@ fsqueue_envelope(enum queue_kind qkind, enum queue_op qop, struct envelope *m)
 
 	return 0;
 }
+
+#define	QWALK_AGAIN	0x1
+#define	QWALK_RECURSE	0x2
+#define	QWALK_RETURN	0x3
+
+struct qwalk {
+	enum queue_kind kind;
+	char	  path[MAXPATHLEN];
+	DIR	 *dirs[3];
+	int	(*filefn)(struct qwalk *, char *);
+	int	  bucket;
+	int	  level;
+	int	  strict;
+	u_int32_t msgid;
+};
+
+int		walk_simple(struct qwalk *, char *);
+int		walk_queue(struct qwalk *, char *);
+int		walk_queue_nobucket(struct qwalk *, char *);
+
+void *
+fsqueue_qwalk_new(enum queue_kind kind, u_int32_t msgid)
+{
+	struct qwalk *q;
+
+	q = calloc(1, sizeof(struct qwalk));
+	if (q == NULL)
+		fatal("qwalk_new: calloc");
+
+	strlcpy(q->path, fsqueue_getpath(kind),
+	    sizeof(q->path));
+
+	q->kind = kind;
+	q->level = 0;
+	q->strict = 0;
+	q->filefn = walk_simple;
+	q->msgid = msgid;
+
+	if (q->msgid) {
+		/* force level and bucket */
+		q->bucket = q->msgid & 0xfff;
+		q->level = 2;
+		if (! bsnprintf(q->path, sizeof(q->path), "%s/%03x/%08x/%s",
+			PATH_QUEUE, q->bucket, q->msgid, PATH_ENVELOPES))
+			fatalx("walk_queue: snprintf");
+	}
+
+	if (smtpd_process == PROC_QUEUE || smtpd_process == PROC_RUNNER)
+		q->strict = 1;
+
+	if (kind == Q_QUEUE)
+		q->filefn = walk_queue;
+	if (kind == Q_INCOMING || kind == Q_ENQUEUE || kind == Q_PURGE)
+		q->filefn = walk_queue_nobucket;
+
+	q->dirs[q->level] = opendir(q->path);
+	if (q->dirs[q->level] == NULL)
+		fatal("qwalk_new: opendir");
+
+	return (q);
+}
+
+int
+fsqueue_qwalk(void *hdl, u_int64_t *evpid)
+{
+	struct qwalk *q = hdl;
+	struct dirent	*dp;
+
+again:
+	errno = 0;
+	dp = readdir(q->dirs[q->level]);
+	if (errno)
+		fatal("qwalk: readdir");
+	if (dp == NULL) {
+		closedir(q->dirs[q->level]);
+		q->dirs[q->level] = NULL;
+		if (q->level == 0 || q->msgid)
+			return (0);
+		q->level--;
+		goto again;
+	}
+
+	if (strcmp(dp->d_name, ".") == 0 || strcmp(dp->d_name, "..") == 0)
+		goto again;
+
+	switch (q->filefn(q, dp->d_name)) {
+	case QWALK_AGAIN:
+		goto again;
+	case QWALK_RECURSE:
+		goto recurse;
+	case QWALK_RETURN: {
+		char *endptr;
+
+		errno = 0;
+		*evpid = (u_int64_t)strtoull(dp->d_name, &endptr, 16);
+		if (q->path[0] == '\0' || *endptr != '\0')
+			goto again;
+		if (errno == ERANGE && *evpid == ULLONG_MAX)
+			goto again;
+		if (q->msgid)
+			if (evpid_to_msgid(*evpid) != q->msgid)
+				return 0;
+
+		return (1);
+	}
+	default:
+		fatalx("qwalk: callback failed");
+	}
+
+recurse:
+	q->level++;
+	q->dirs[q->level] = opendir(q->path);
+	if (q->dirs[q->level] == NULL) {
+		if (errno == ENOENT && !q->strict) {
+			q->level--;
+			goto again;
+		}
+		fatal("qwalk: opendir");
+	}
+	goto again;
+}
+
+void
+fsqueue_qwalk_close(void *hdl)
+{
+	int i;
+	struct qwalk *q = hdl;
+
+	for (i = 0; i <= q->level; i++)
+		if (q->dirs[i])
+			closedir(q->dirs[i]);
+
+	bzero(q, sizeof(struct qwalk));
+	free(q);
+}
+
+
+int
+walk_simple(struct qwalk *q, char *fname)
+{
+	return (QWALK_RETURN);
+}
+
+int
+walk_queue(struct qwalk *q, char *fname)
+{
+	char	*ep;
+
+	switch (q->level) {
+	case 0:
+		if (strcmp(fname, "envelope.tmp") == 0)
+			return (QWALK_AGAIN);
+
+		q->bucket = strtoul(fname, &ep, 16);
+		if (fname[0] == '\0' || *ep != '\0') {
+			log_warnx("walk_queue: invalid bucket: %s", fname);
+			return (QWALK_AGAIN);
+		}
+		if (errno == ERANGE || q->bucket >= DIRHASH_BUCKETS) {
+			log_warnx("walk_queue: invalid bucket: %s", fname);
+			return (QWALK_AGAIN);
+		}
+		if (! bsnprintf(q->path, sizeof(q->path), "%s/%03x",
+			fsqueue_getpath(q->kind), q->bucket & 0xfff))
+			fatalx("walk_queue: snprintf");
+		return (QWALK_RECURSE);
+	case 1:
+		if (! bsnprintf(q->path, sizeof(q->path), "%s/%03x/%s%s",
+			fsqueue_getpath(q->kind), q->bucket & 0xfff, fname,
+			PATH_ENVELOPES))
+			fatalx("walk_queue: snprintf");
+		return (QWALK_RECURSE);
+	case 2:
+		return (QWALK_RETURN);
+	}
+
+	return (-1);
+}
+
+int
+walk_queue_nobucket(struct qwalk *q, char *fname)
+{
+	switch (q->level) {
+	case 0:
+		if (strcmp(fname, "envelope.tmp") == 0)
+			return (QWALK_AGAIN);
+		if (! bsnprintf(q->path, sizeof(q->path), "%s/%s%s",
+			fsqueue_getpath(q->kind), fname, PATH_ENVELOPES))
+			fatalx("walk_queue_nobucket: snprintf");
+		return (QWALK_RECURSE);
+	case 1:
+		return (QWALK_RETURN);
+	}
+
+	return (-1);
+}
diff --git a/usr.sbin/smtpd/queue_shared.c b/usr.sbin/smtpd/queue_shared.c
index fd32686696f..515a6f521e5 100644
--- a/usr.sbin/smtpd/queue_shared.c
+++ b/usr.sbin/smtpd/queue_shared.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: queue_shared.c,v 1.53 2011/10/27 14:32:57 chl Exp $	*/
+/*	$OpenBSD: queue_shared.c,v 1.54 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -40,24 +40,8 @@
 #include "smtpd.h"
 #include "log.h"
 
-#define	QWALK_AGAIN	0x1
-#define	QWALK_RECURSE	0x2
-#define	QWALK_RETURN	0x3
-
 int	fsqueue_load_envelope_ascii(FILE *, struct envelope *);
 
-struct qwalk {
-	char	  path[MAXPATHLEN];
-	DIR	 *dirs[3];
-	int	(*filefn)(struct qwalk *, char *);
-	int	  bucket;
-	int	  level;
-	int	  strict;
-};
-
-int		walk_simple(struct qwalk *, char *);
-int		walk_queue(struct qwalk *, char *);
-
 void		display_envelope(struct envelope *, int);
 void		getflag(u_int *, int, char *, char *, size_t);
 
@@ -117,166 +101,24 @@ queue_message_update(struct envelope *e)
 	queue_envelope_delete(Q_QUEUE, e);
 }
 
-struct qwalk *
-qwalk_new(char *path)
-{
-	struct qwalk *q;
-
-	q = calloc(1, sizeof(struct qwalk));
-	if (q == NULL)
-		fatal("qwalk_new: calloc");
-
-	strlcpy(q->path, path, sizeof(q->path));
-
-	q->level = 0;
-	q->strict = 0;
-	q->filefn = walk_simple;
-
-	if (smtpd_process == PROC_QUEUE || smtpd_process == PROC_RUNNER)
-		q->strict = 1;
-
-	if (strcmp(path, PATH_QUEUE) == 0)
-		q->filefn = walk_queue;
-
-	q->dirs[0] = opendir(q->path);
-	if (q->dirs[0] == NULL)
-		fatal("qwalk_new: opendir");
-
-	return (q);
-}
-
-int
-qwalk(struct qwalk *q, char *filepath)
-{
-	struct dirent	*dp;
-
-again:
-	errno = 0;
-	dp = readdir(q->dirs[q->level]);
-	if (errno)
-		fatal("qwalk: readdir");
-	if (dp == NULL) {
-		closedir(q->dirs[q->level]);
-		q->dirs[q->level] = NULL;
-		if (q->level == 0)
-			return (0);
-		q->level--;
-		goto again;
-	}
-
-	if (strcmp(dp->d_name, ".") == 0 || strcmp(dp->d_name, "..") == 0)
-		goto again;
-
-	switch (q->filefn(q, dp->d_name)) {
-	case QWALK_AGAIN:
-		goto again;
-	case QWALK_RECURSE:
-		goto recurse;
-	case QWALK_RETURN:
-		if (! bsnprintf(filepath, MAXPATHLEN, "%s/%s", q->path,
-			dp->d_name))
-			fatalx("qwalk: snprintf");
-		return (1);
-	default:
-		fatalx("qwalk: callback failed");
-	}
-
-recurse:
-	q->level++;
-	q->dirs[q->level] = opendir(q->path);
-	if (q->dirs[q->level] == NULL) {
-		if (errno == ENOENT && !q->strict) {
-			q->level--;
-			goto again;
-		}
-		fatal("qwalk: opendir");
-	}
-	goto again;
-}
-
 void
-qwalk_close(struct qwalk *q)
+show_queue(enum queue_kind kind, int flags)
 {
-	int i;
-
-	for (i = 0; i <= q->level; i++)
-		if (q->dirs[i])
-			closedir(q->dirs[i]);
-
-	bzero(q, sizeof(struct qwalk));
-	free(q);
-}
-
-int
-walk_simple(struct qwalk *q, char *fname)
-{
-	return (QWALK_RETURN);
-}
-
-int
-walk_queue(struct qwalk *q, char *fname)
-{
-	char	*ep;
-
-	switch (q->level) {
-	case 0:
-		if (strcmp(fname, "envelope.tmp") == 0)
-			return (QWALK_AGAIN);
-
-		q->bucket = strtoul(fname, &ep, 16);
-		if (fname[0] == '\0' || *ep != '\0') {
-			log_warnx("walk_queue: invalid bucket: %s", fname);
-			return (QWALK_AGAIN);
-		}
-		if (errno == ERANGE || q->bucket >= DIRHASH_BUCKETS) {
-			log_warnx("walk_queue: invalid bucket: %s", fname);
-			return (QWALK_AGAIN);
-		}
-		if (! bsnprintf(q->path, sizeof(q->path), "%s/%03x", PATH_QUEUE,
-			q->bucket & 0xfff))
-			fatalx("walk_queue: snprintf");
-		return (QWALK_RECURSE);
-	case 1:
-		if (! bsnprintf(q->path, sizeof(q->path), "%s/%03x/%s%s",
-			PATH_QUEUE, q->bucket & 0xfff, fname, PATH_ENVELOPES))
-			fatalx("walk_queue: snprintf");
-		return (QWALK_RECURSE);
-	case 2:
-		return (QWALK_RETURN);
-	}
-
-	return (-1);
-}
-
-void
-show_queue(char *queuepath, int flags)
-{
-	char		 path[MAXPATHLEN];
-	struct envelope	 message;
 	struct qwalk	*q;
-	FILE		*fp;
+	struct envelope	 envelope;
+	u_int64_t	 evpid;
 
 	log_init(1);
 
 	if (chroot(PATH_SPOOL) == -1 || chdir(".") == -1)
 		err(1, "%s", PATH_SPOOL);
 
-	q = qwalk_new(queuepath);
-
-	while (qwalk(q, path)) {
-		fp = fopen(path, "r");
-		if (fp == NULL) {
-			if (errno == ENOENT)
-				continue;
-			err(1, "%s", path);
-		}
-
-		errno = 0;
-		if (! fsqueue_load_envelope_ascii(fp, &message))
-			err(1, "%s", path);
-		fclose(fp);
+	q = qwalk_new(kind, 0);
 
-		display_envelope(&message, flags);
+	while (qwalk(q, &evpid)) {
+		if (! queue_envelope_load(kind, evpid, &envelope))
+			continue;
+		display_envelope(&envelope, flags);
 	}
 
 	qwalk_close(q);
diff --git a/usr.sbin/smtpd/ramqueue.c b/usr.sbin/smtpd/ramqueue.c
index 570b0cf62fd..f1568cae23e 100644
--- a/usr.sbin/smtpd/ramqueue.c
+++ b/usr.sbin/smtpd/ramqueue.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: ramqueue.c,v 1.25 2011/11/07 11:14:10 eric Exp $	*/
+/*	$OpenBSD: ramqueue.c,v 1.26 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2011 Gilles Chehade <gilles@openbsd.org>
@@ -112,26 +112,21 @@ ramqueue_batch_first_envelope(struct ramqueue_batch *rq_batch)
 int
 ramqueue_load(struct ramqueue *rqueue, time_t *nsched)
 {
-	char			path[MAXPATHLEN];
+//	char			path[MAXPATHLEN];
 	time_t			curtm;
 	struct envelope		envelope;
 	static struct qwalk    *q = NULL;
 	struct ramqueue_envelope *rq_evp;
-
-
+//	u_int32_t	msgid;
+	u_int64_t	evpid;
 
 	log_debug("ramqueue: queue loading in progress");
 
 	if (q == NULL)
-		q = qwalk_new(PATH_QUEUE);
-	while (qwalk(q, path)) {
-		u_int64_t evpid;
+		q = qwalk_new(Q_QUEUE, 0);
 
+	while (qwalk(q, &evpid)) {
 		curtm = time(NULL);
-
-		if ((evpid = filename_to_evpid(basename(path))) == 0)
-			continue;
-
 		if (! queue_envelope_load(Q_QUEUE, evpid, &envelope)) {
 			log_debug("failed to load envelope");
 			queue_message_corrupt(Q_QUEUE, evpid_to_msgid(evpid));
diff --git a/usr.sbin/smtpd/runner.c b/usr.sbin/smtpd/runner.c
index ec689d279c1..231241f60e9 100644
--- a/usr.sbin/smtpd/runner.c
+++ b/usr.sbin/smtpd/runner.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: runner.c,v 1.124 2011/11/14 19:23:41 chl Exp $	*/
+/*	$OpenBSD: runner.c,v 1.125 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -55,7 +55,6 @@ static void runner_timeout(int, short, void *);
 static int runner_process_envelope(struct ramqueue_envelope *, time_t);
 static void runner_process_batch(struct ramqueue_envelope *, time_t);
 static void runner_purge_run(void);
-static void runner_purge_message(u_int32_t);
 static int runner_check_loop(struct envelope *);
 static int runner_force_message_to_ramqueue(struct ramqueue *, u_int32_t);
 
@@ -285,8 +284,6 @@ runner(void)
 	config_pipes(peers, nitems(peers));
 	config_peers(peers, nitems(peers));
 
-	unlink(PATH_QUEUE "/envelope.tmp");
-
 	runner_setup_events();
 	event_dispatch();
 	runner_disable_events();
@@ -492,43 +489,23 @@ runner_process_batch(struct ramqueue_envelope *rq_evp, time_t curtm)
 		ramqueue_remove_host(&env->sc_rqueue, rq_host);
 }
 
-/* XXX - temporary solution */
 int
 runner_force_message_to_ramqueue(struct ramqueue *rqueue, u_int32_t msgid)
 {
-	char path[MAXPATHLEN];
-	DIR *dirp;
-	struct dirent *dp;
-	struct envelope envelope;
-	time_t curtm;
-
-	if (! bsnprintf(path, MAXPATHLEN, "%s/%03x/%08x/envelopes",
-		PATH_QUEUE, msgid & 0xfff, msgid))
-		return 0;
-
-	dirp = opendir(path);
-	if (dirp == NULL)
-		return 0;
+	struct qwalk	*q;
+	u_int64_t	 evpid;
+	time_t		 curtm;
+	struct envelope	 envelope;
 
 	curtm = time(NULL);
-	while ((dp = readdir(dirp)) != NULL) {
-		u_int64_t evpid;
-
-		if (dp->d_name[0] == '.')
-			continue;
-
-		if ((evpid = filename_to_evpid(dp->d_name)) == 0) {
-			log_warnx("runner_force_message_to_ramqueue: "
-				  "invalid evpid: %016" PRIx64, evpid);
-			continue;
-		}
-
+	q = qwalk_new(Q_QUEUE, msgid);
+	while (qwalk(q, &evpid)) {
 		if (! queue_envelope_load(Q_QUEUE, evpid,
 			&envelope))
 			continue;
 		ramqueue_insert(rqueue, &envelope, curtm);
 	}
-	closedir(dirp);
+ 	qwalk_close(q);
 
 	return 1;
 }
@@ -536,83 +513,18 @@ runner_force_message_to_ramqueue(struct ramqueue *rqueue, u_int32_t msgid)
 void
 runner_purge_run(void)
 {
-	char		 path[MAXPATHLEN];
 	struct qwalk	*q;
+	u_int32_t	 msgid;
+	u_int64_t	 evpid;
 
-	q = qwalk_new(PATH_PURGE);
-
-	while (qwalk(q, path)) {
-		u_int32_t msgid;
-		char *bpath;
-		
-		bpath = basename(path);
-		if (bpath[0] == '.')
-			continue;
-
-		if ((msgid = filename_to_msgid(bpath)) == 0) {
-			log_warnx("runner_purge_run: invalid msgid: %08x", msgid);
-			continue;
-		}
-		runner_purge_message(msgid);
+	q = qwalk_new(Q_PURGE, 0);
+	while (qwalk(q, &evpid)) {
+		msgid = evpid_to_msgid(evpid);
+		queue_message_delete(Q_PURGE, msgid);
 	}
-
 	qwalk_close(q);
 }
 
-void
-runner_purge_message(u_int32_t msgid)
-{
-	char rootdir[MAXPATHLEN];
-	char evpdir[MAXPATHLEN];
-	char evppath[MAXPATHLEN];
-	char msgpath[MAXPATHLEN];
-	DIR *dirp;
-	struct dirent *dp;
-	
-	if (! bsnprintf(rootdir, sizeof(rootdir), "%s/%08x", PATH_PURGE, msgid))
-		fatal("runner_purge_message: snprintf");
-
-	if (! bsnprintf(evpdir, sizeof(evpdir), "%s%s", rootdir,
-		PATH_ENVELOPES))
-		fatal("runner_purge_message: snprintf");
-	
-	if (! bsnprintf(msgpath, sizeof(msgpath), "%s/message", rootdir))
-		fatal("runner_purge_message: snprintf");
-
-	if (unlink(msgpath) == -1)
-		if (errno != ENOENT)
-			fatal("runner_purge_message: unlink");
-
-	dirp = opendir(evpdir);
-	if (dirp == NULL) {
-		if (errno == ENOENT)
-			goto delroot;
-		fatal("runner_purge_message: opendir");
-	}
-	while ((dp = readdir(dirp)) != NULL) {
-		if (strcmp(dp->d_name, ".") == 0 ||
-		    strcmp(dp->d_name, "..") == 0)
-			continue;
-		if (! bsnprintf(evppath, sizeof(evppath), "%s/%s", evpdir,
-			dp->d_name))
-			fatal("runner_purge_message: snprintf");
-
-		if (unlink(evppath) == -1)
-			if (errno != ENOENT)
-				fatal("runner_purge_message: unlink");
-	}
-	closedir(dirp);
-
-	if (rmdir(evpdir) == -1)
-		if (errno != ENOENT)
-			fatal("runner_purge_message: rmdir");
-
-delroot:
-	if (rmdir(rootdir) == -1)
-		if (errno != ENOENT)
-			fatal("runner_purge_message: rmdir");
-}
-
 int
 runner_check_loop(struct envelope *ep)
 {
diff --git a/usr.sbin/smtpd/smtpctl.c b/usr.sbin/smtpd/smtpctl.c
index 654f47ff15c..f71c96000c6 100644
--- a/usr.sbin/smtpd/smtpctl.c
+++ b/usr.sbin/smtpd/smtpctl.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpctl.c,v 1.70 2011/10/26 20:47:31 gilles Exp $	*/
+/*	$OpenBSD: smtpctl.c,v 1.71 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2006 Pierre-Yves Ritschard <pyr@openbsd.org>
@@ -30,6 +30,7 @@
 #include <errno.h>
 #include <event.h>
 #include <imsg.h>
+#include <pwd.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -81,20 +82,34 @@ main(int argc, char *argv[])
 	else if (strcmp(__progname, "mailq") == 0) {
 		if (geteuid())
 			errx(1, "need root privileges");
-		show_queue(PATH_QUEUE, 0);
+		show_queue(Q_QUEUE, 0);
 		return 0;
 	} else if (strcmp(__progname, "smtpctl") == 0) {
+		struct smtpd smtpd;
+
 		/* check for root privileges */
 		if (geteuid())
 			errx(1, "need root privileges");
 
+		bzero(&smtpd, sizeof (smtpd));
+		env = &smtpd;
+		if ((env->sc_pw = getpwnam(SMTPD_USER)) == NULL)
+			errx(1, "unknown user %s", SMTPD_USER);
+
+		env->sc_queue = queue_backend_lookup(QT_FS);
+		if (env->sc_queue == NULL)
+			errx(1, "could not find queue backend");
+
+		if (!env->sc_queue->init())
+			errx(1, "invalid directory permissions");
+
 		if ((res = parse(argc - 1, argv + 1)) == NULL)
 			exit(1);
 
 		/* handle "disconnected" commands */
 		switch (res->action) {
 		case SHOW_QUEUE:
-			show_queue(PATH_QUEUE, 0);
+			show_queue(Q_QUEUE, 0);
 			break;
 		case SHOW_RUNQUEUE:
 			break;
diff --git a/usr.sbin/smtpd/smtpd.h b/usr.sbin/smtpd/smtpd.h
index 57f72ae1191..701822053e9 100644
--- a/usr.sbin/smtpd/smtpd.h
+++ b/usr.sbin/smtpd/smtpd.h
@@ -1,4 +1,4 @@
-/*	$OpenBSD: smtpd.h,v 1.250 2011/11/14 19:23:41 chl Exp $	*/
+/*	$OpenBSD: smtpd.h,v 1.251 2011/11/15 23:06:39 gilles Exp $	*/
 
 /*
  * Copyright (c) 2008 Gilles Chehade <gilles@openbsd.org>
@@ -64,18 +64,7 @@
 #define	DIRHASH_BUCKETS		 4096
 
 #define PATH_SPOOL		"/var/spool/smtpd"
-
-#define PATH_ENQUEUE		"/enqueue"
-#define PATH_INCOMING		"/incoming"
-#define PATH_QUEUE		"/queue"
-#define PATH_PURGE		"/purge"
-#define PATH_CORRUPT		"/corrupt"
-
-#define PATH_MESSAGE		"/message"
-#define PATH_ENVELOPES		"/envelopes"
-
 #define PATH_OFFLINE		"/offline"
-#define PATH_BOUNCE		"/bounce"
 
 /* number of MX records to lookup */
 #define MAX_MX_COUNT		10
@@ -898,7 +887,6 @@ enum queue_kind {
 	Q_INCOMING,
 	Q_QUEUE,
 	Q_PURGE,
-	Q_OFFLINE,
 	Q_BOUNCE,
 	Q_CORRUPT
 };
@@ -918,9 +906,14 @@ enum queue_op {
 
 struct queue_backend {
 	enum queue_type	type;
+
 	int (*init)(void);
 	int (*message)(enum queue_kind, enum queue_op, u_int32_t *);
 	int (*envelope)(enum queue_kind, enum queue_op, struct envelope *);
+
+	void *(*qwalk_new)(enum queue_kind, u_int32_t);
+	int   (*qwalk)(void *, u_int64_t *);
+	void  (*qwalk_close)(void *);
 };
 
 
@@ -1073,6 +1066,8 @@ void queue_commit_envelopes(struct envelope *);
 
 
 /* queue_backend.c */
+u_int32_t queue_generate_msgid(void);
+u_int64_t queue_generate_evpid(u_int32_t msgid);
 struct queue_backend *queue_backend_lookup(enum queue_type);
 int queue_message_create(enum queue_kind, u_int32_t *);
 int queue_message_delete(enum queue_kind, u_int32_t);
@@ -1085,15 +1080,15 @@ int queue_envelope_create(enum queue_kind, struct envelope *);
 int queue_envelope_delete(enum queue_kind, struct envelope *);
 int queue_envelope_load(enum queue_kind, u_int64_t, struct envelope *);
 int queue_envelope_update(enum queue_kind, struct envelope *);
+void *qwalk_new(enum queue_kind, u_int32_t);
+int   qwalk(void *, u_int64_t *);
+void  qwalk_close(void *);
 
 
 /* queue_shared.c */
 void queue_message_update(struct envelope *);
-struct qwalk	*qwalk_new(char *);
-int qwalk(struct qwalk *, char *);
-void qwalk_close(struct qwalk *);
 int bounce_record_message(struct envelope *, struct envelope *);
-void show_queue(char *, int);
+void show_queue(enum queue_kind, int);
 
 
 /* ramqueue.c */