From: Simon Schubert <2@0x2c.org>
Date: Sat, 18 Feb 2012 00:15:22 +0000 (+0100)
Subject: implement queue flushing prod
X-Git-Url: http://git.ipfire.org/?p=people%2Fms%2Fdma.git;a=commitdiff_plain;h=3686aecc4a2ebd388e0879ff069ee6bf42f3e50e

implement queue flushing prod

So far dma would sleep for the whole backoff period before retrying a
delivery.  However, this also meant that a user issuing `dma -q` could
not force these mails to be retried immediately.

Switch to sleeping for only 30 seconds at a time, and poll the mtime
of a special queue flush signal file `flush` in the spool directory.
If the flush file has been touched since the beginning of the backoff
period, initiate a retry right away.

When flushing the queue with `dma -q` we now touch the flush file to
make waiting instances of dma process this user request.

The 30 second poll interval at the same time avoids a DoS via
excessive `dma -q` by any user.

fixes #3
Bug: https://github.com/corecode/dma/issues/3
---

diff --git a/dma.c b/dma.c
index c9508f2..bb5fa19 100644
--- a/dma.c
+++ b/dma.c
@@ -302,7 +302,7 @@ static void
 deliver(struct qitem *it)
 {
 	int error;
-	unsigned int backoff = MIN_RETRY;
+	unsigned int backoff = MIN_RETRY, slept;
 	struct timeval now;
 	struct stat st;
 
@@ -334,7 +334,14 @@ retry:
 				 MAX_TIMEOUT);
 			goto bounce;
 		}
-		if (sleep(backoff) == 0) {
+		for (slept = 0; slept < backoff;) {
+			slept += SLEEP_TIMEOUT - sleep(SLEEP_TIMEOUT);
+			if (flushqueue_since(slept)) {
+				backoff = MIN_RETRY;
+				goto retry;
+			}
+		}
+		if (slept >= backoff) {
 			/* pick the next backoff between [1.5, 2.5) times backoff */
 			backoff = backoff + backoff / 2 + random() % backoff;
 			if (backoff > MAX_RETRY)
@@ -554,6 +561,7 @@ skipopts:
 	}
 
 	if (doqueue) {
+		flushqueue_signal();
 		if (load_queue(&queue) < 0)
 			errlog(1, "can not load queue");
 		run_queue(&queue);
diff --git a/dma.h b/dma.h
index 1dae38c..440a7a3 100644
--- a/dma.h
+++ b/dma.h
@@ -52,6 +52,7 @@
 #define MIN_RETRY	300		/* 5 minutes */
 #define MAX_RETRY	(3*60*60)	/* retry at least every 3 hours */
 #define MAX_TIMEOUT	(5*24*60*60)	/* give up after 5 days */
+#define SLEEP_TIMEOUT	30		/* check for queue flush every 30 seconds */
 #ifndef PATH_MAX
 #define PATH_MAX	1024		/* Max path len */
 #endif
@@ -74,6 +75,8 @@
 #error Please define LIBEXEC_PATH
 #endif
 
+#define SPOOL_FLUSHFILE	"flush"
+
 #define DMA_ROOT_USER	"mail"
 #define DMA_GROUP	"mail"
 
@@ -203,6 +206,8 @@ int load_queue(struct queue *);
 void delqueue(struct qitem *);
 int acquirespool(struct qitem *);
 void dropspool(struct queue *, struct qitem *);
+int flushqueue_since(unsigned int);
+int flushqueue_signal(void);
 
 /* local.c */
 int deliver_local(struct qitem *);
diff --git a/spool.c b/spool.c
index 7c7c08d..873352d 100644
--- a/spool.c
+++ b/spool.c
@@ -395,3 +395,52 @@ dropspool(struct queue *queue, struct qitem *keep)
 			fclose(it->mailf);
 	}
 }
+
+int
+flushqueue_since(unsigned int period)
+{
+        struct stat st;
+	struct timeval now;
+        char *flushfn = NULL;
+
+	if (asprintf(&flushfn, "%s/%s", config.spooldir, SPOOL_FLUSHFILE) < 0)
+		return (0);
+	if (stat(flushfn, &st) < 0) {
+		free(flushfn);
+		return (0);
+	}
+	free(flushfn);
+	flushfn = NULL;
+	if (gettimeofday(&now, 0) != 0)
+		return (0);
+
+	/* Did the flush file get touched within the last period seconds? */
+	if (st.st_mtim.tv_sec + period >= now.tv_sec)
+		return (1);
+	else
+		return (0);
+}
+
+int
+flushqueue_signal(void)
+{
+        char *flushfn = NULL;
+	int fd;
+
+        if (asprintf(&flushfn, "%s/%s", config.spooldir, SPOOL_FLUSHFILE) < 0)
+		return (-1);
+	fd = open(flushfn, O_CREAT|O_RDONLY, 0440);
+	if (fd < 0) {
+		syslog(LOG_ERR, "could not open flush file: %m");
+		free(flushfn);
+		return (-1);
+	}
+        close(fd);
+        if (utimes(flushfn, NULL) < 0) {
+		syslog(LOG_ERR, "could not touch flush file: %m");
+		free(flushfn);
+		return (-1);
+	}
+	free (flushfn);
+	return (0);
+}