Deal properly with backlogs when feed dropped

diff --git a/backends/innduct.c b/backends/innduct.c
index 1c41554258d8b1d6e985a274a427262b26ba42c6..57eb9dba3e99e11b76ad94f4e19c8285038dfad3 100644 (file)
--- a/backends/innduct.c
+++ b/backends/innduct.c
@@ -1393,9 +1393,9 @@ static void inputfile_tailing_stop(InputFile *ipf) {
            poll for F
           ================
                |
-               |     TIMEOUT
+               |     TIMEOUT and no defer, no backlog
                |`--------------------------.
-               |                           | install defer as backlog
+               |                           |
                 | OPEN F SUCCEEDS           | exit
      ,--------->|                           V
      |          V                         =========
@@ -1504,13 +1504,17 @@ static void statemc_init(void) {
   }
   debug("startup: locked");
 
+  search_backlog_file();
+
   xlstat_isreg(path_ductdefer, &stab, &noent, "defer file");
   if (noent) {
     debug("startup: ductdefer ENOENT");
   } else {
     debug("startup: ductdefer nlink=%ld", (long)stab.st_nlink);
     switch (stab.st_nlink==1) {
-    case 1: /* ok */ break;
+    case 1:
+      open_defer(); /* so that we will later close it and rename it */
+      break;
     case 2:
       if (unlink(path_defer))
        sysdie("could not unlink stale defer file link %s (presumably"
@@ -1599,8 +1603,13 @@ static void *statemc_check_input_done(oop_source *lp,
   if (ipf == backlog_input_file) {
     notice_processed(ipf,"backlog file",ipf->path);
     close_input_file(ipf);
-    if (unlink(ipf->path))
-      sysdie("could not unlink done backlog file %s", ipf->path);
+    if (unlink(ipf->path)) {
+      if (errno != ENOENT)
+       sysdie("could not unlink processed backlog file %s", ipf->path);
+      warn("backlog file %s vanished while we were reading it"
+          " so we couldn't remove it (but it's done now, anyway)",
+          ipf->path);
+    }
     backlog_input_file= 0;
     search_backlog_file();
     return;
@@ -1617,8 +1626,11 @@ static void *statemc_check_input_done(oop_source *lp,
     sysdie("could not unlink old duct file %s", path_duct);
 
   if (sms==sm_DROPPING) {
-    notice("feed dropped and our work is complete"
-          " (but check for backlog files)");
+    if (search_backlog_file()) {
+      debug("feed dropped but still backlogs to process");
+      return;
+    }
+    notice("feed dropped and our work is complete");
     r= unlink(path_ductlock);
     if (r) sysdie("unlink lock file for old feed %s", path_ductlock);
     exit(0);
@@ -1730,14 +1742,18 @@ static void poll_backlog_file(void) {
   search_backlog_file();
 }
 
-static void search_backlog_file(void) {
+static int search_backlog_file(void) {
+  /* returns non-0 iff there are any backlog files */
+
   glob_t gl;
   int r;
   struct stat stab;
   const char *oldest_path=0;
   time_t oldest_mtime, now;
 
-  assert(!backlog_input_file);
+  if (backlog_input_file) return 3;
+
+ try_again:
 
   r= glob(globpat_backlog, GLOB_ERR|GLOB_MARK|GLOB_NOSORT, 0, &gl);
 
@@ -1775,7 +1791,7 @@ static void search_backlog_file(void) {
   if (!oldest_path) {
     debug("backlog scan: none");
     backlog_nextscan_periods= backlog_spontaneous_rescan_periods;
-    return;
+    return 0;
   }
 
   now= time();  if (now==-1) sysdie("time(2) failed");
@@ -1787,9 +1803,13 @@ static void search_backlog_file(void) {
          age, age_deficiency, oldest_path);
 
     backlog_input_file= open_input_file();
+    if (!backlog_input_file) {
+      warn("backlog file %s vanished as we opened it", backlog_input_file);
+      goto try_again;
+    }
     inputfile_tailing_start(backlog_input_file);
     backlog_nextscan_periods= -1;
-    return;
+    return 1;
   }
 
   backlog_nextscan_periods= age_deficiency / PERIOD_SECONDS;
@@ -1800,6 +1820,7 @@ static void search_backlog_file(void) {
 
   debug("backlog scan: young age=%f deficiency=%ld nextscan=%d oldest=%s",
        age, age_deficiency, backlog_nextscan_periods, oldest_path);
+  return 2;
 }
 
 /*========== flushing the feed ==========*/