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 c9508f227ab621f39fde89f36e86e8c800831cbe..bb5fa198f18adc6c039a80097aa9ebcdd35369c5 100644 (file)
--- 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 1dae38cb20d4885b2de25577b0153dc9fe63aca1..440a7a315eb2cd33f19693c749d4d0f81d3da95e 100644 (file)
--- 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 7c7c08d5c53503ac56c8e43aa4f1741b0f9854b6..873352d4be2fc7742bb27f9f60be5a406174651c 100644 (file)
--- 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);
+}