+ if (fflush(defer))
+ sysfatal("could not flush defer file %s", path_defer);
+ if (ftruncate(fileno(defer), truncto))
+ sysdie("could not truncate defer file %s", path_defer);
+
+ } else {
+ info("continuing existing defer file %s (%ld bytes)",
+ path_defer, orgsize);
+ }
+ if (fseek(defer, truncto, SEEK_SET))
+ sysdie("could not seek to new end of defer file %s", path_defer);
+}
+
+static void close_defer(void) {
+ if (!defer)
+ return;
+
+ xfstat(fileno(defer), &stab, "defer file");
+
+ if (fclose(defer)) sysfatal("could not close defer file %s", path_defer);
+ defer= 0;
+
+ char *backlog= xasprintf("%s_backlog_%lu.%lu", feedfile,
+ (unsigned long)now.tv_sec,
+ (unsigned long)stab.st_ino);
+ if (link(path_defer, path_backlog))
+ sysfatal("could not install defer file %s as backlog file %s",
+ path_defer, backlog);
+ if (unlink(path_defer))
+ sysdie("could not unlink old defer link %s to backlog file %s",
+ path_defer, backlog);
+
+ if (until_backlog_nextscan < 0 ||
+ until_backlog_nextscan > backlog_retry_minperiods + 1)
+ until_backlog_nextscan= backlog_retry_minperiods + 1;
+}
+
+static void poll_backlog_file(void) {
+ if (until_backlog_nextscan < 0) return;
+ if (until_backlog_nextscan-- > 0) return;
+ search_backlog_file();
+}
+
+static void 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;
+
+ if (backlog_input_file) return 3;
+
+ try_again:
+
+ r= glob(globpat_backlog, GLOB_ERR|GLOB_MARK|GLOB_NOSORT, 0, &gl);
+
+ switch (r) {
+ case GLOB_ABORTED:
+ sysdie("failed to expand backlog pattern %s", globpat_backlog);
+ case GLOB_NOSPACE:
+ die("out of memory expanding backlog pattern %s", globpat_backlog);
+ case 0:
+ for (i=0; i<gl.gl_pathc; i++) {
+ const char *path= gl.gl_pathv[i];
+
+ if (strchr(path,'#') || strchr(path,'~')) {
+ debug("backlog file search skipping %s", path);
+ continue;
+ }
+ r= stat(path, &stab);
+ if (r) {
+ syswarn("failed to stat backlog file %s", path);
+ continue;
+ }
+ if (!S_ISREG(stab.st_mode)) {
+ warn("backlog file %s is not a plain file (or link to one)", path);
+ continue;
+ }
+ if (!oldest_path || stab.st_mtime < oldest_mtime) {
+ oldest_path= path;
+ oldest_mtime= stab.st_mtime;
+ }
+ }
+ case GLOB_NOMATCH: /* fall through */
+ break;
+ default:
+ sysdie("glob expansion of backlog pattern %s gave unexpected"
+ " nonzero (error?) return value %d", globpat_backlog, r);
+ }
+
+ globfree(&gl);
+
+ if (!oldest_path) {
+ debug("backlog scan: none");
+
+ if (sms==sm_DROPPED) {
+ notice("feed dropped and our work is complete");
+ xunlink(path_lock, "lockfile for old feed");
+ exit(0);
+ }
+ until_backlog_nextscan= backlog_spontaneous_rescan_periods;
+ return 0;
+ }
+
+ now= time(); if (now==-1) sysdie("time(2) failed");
+ double age= difftime(now, oldest_mtime);
+ long age_deficiency= (backlog_retry_minperiods * PERIOD_SECONDS) - age;
+
+ if (age_deficiency <= 0) {
+ debug("backlog scan: found age=%f deficiency=%ld oldest=%s",
+ age, age_deficiency, oldest_path);
+
+ backlog_input_file= open_input_file(oldest_path);
+ 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);
+ until_backlog_nextscan= -1;
+ return 1;
+ }
+
+ until_backlog_nextscan= age_deficiency / PERIOD_SECONDS;
+
+ if (backlog_spontaneous_rescan_periods >= 0 &&
+ until_backlog_nextscan > backlog_spontaneous_rescan_periods)
+ until_backlog_nextscan= backlog_spontaneous_rescan_periods;
+
+ debug("backlog scan: young age=%f deficiency=%ld nextscan=%d oldest=%s",
+ age, age_deficiency, until_backlog_nextscan, oldest_path);
+ return 2;
+}
+
+/*========== flushing the feed ==========*/
+
+static pid_t inndcomm_child;
+
+static void *inndcomm_event(oop_source *lp, int fd, oop_event e, void *u) {
+ assert(inndcomm_child);
+ int status= xwaitpid(&inndcomm_child, "inndcomm");
+ loop->cancel_fd(fd);
+ close(fd);
+
+ assert(!flushing_input_file);
+
+ if (WIFEXITED(status)) {
+ switch (WEXITSTATUS(status)) {
+
+ case INNDCOMMCHILD_ESTATUS_FAIL:
+ goto failed;
+
+ case INNDCOMMCHILD_ESTATUS_NONESUCH:
+ warn("feed has been dropped by innd, finishing up");
+ flushing_input_file= main_input_file;
+ tailing_queue_readable(flushing_input_file);
+ /* we probably previously returned EAGAIN from our fake read method
+ * when in fact we were at EOF, so signal another readable event
+ * so we actually see the EOF */
+
+ main_input_file= 0;
+
+ if (flushing_input_file) {
+ SMS(DROPPING, 0, "feed dropped by innd, but must finish last flush");
+ } else {
+ close_defer();
+ SMS(DROPPED, 0, "feed dropped by innd");
+ search_backlog_file();
+ }
+ return OOP_CONTINUE;
+
+ case 0:
+ /* as above */
+ flushing_input_file= main_input_file;
+ tailing_queue_readable(flushing_input_file);
+
+ main_input_file= open_input_file(feedfile);
+ if (!main_input_file)
+ die("flush succeeded but feedfile %s does not exist!", feedfile);
+
+ if (flushing_input_file) {
+ SMS(SEPARATED, spontaneous_flush_periods, "recovery flush complete");
+ } else {
+ close_defer();
+ SMS(NORMAL, spontaneous_flush_periods, "flush complete");
+ }
+ return OOP_CONTINUE;
+
+ default:
+ goto unexpected_exitstatus;
+
+ }
+ } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
+ warn("flush timed out trying to talk to innd");
+ goto failed;
+ } else {
+ unexpected_exitstatus:
+ report_child_status("inndcomm child", status);
+ }
+
+ failed:
+ SMS(FLUSHFAILED, flushfail_retry_periods, "flush failed, will retry");
+}
+
+static void inndcommfail(const char *what) {
+ syswarn("error communicating with innd: %s failed: %s", what, ICCfailure);
+ exit(INNDCOMMCHILD_ESTATUS_FAIL);
+}
+
+void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */
+ int pipefds[2];
+
+ notice("flushing %s",why);
+
+ assert(sms==sm_NORMAL || sms==sm_FLUSHFAILED);
+ assert(!inndcomm_child);
+
+ if (pipe(pipefds)) sysdie("create pipe for inndcomm child sentinel");
+
+ inndcomm_child= xfork();
+
+ if (!inndcomm_child) {
+ static char flushargv[2]= { sitename, 0 };
+ char *reply;
+
+ close(pipefds[0]);
+
+ alarm(inndcomm_flush_timeout);
+ r= ICCopen(); if (r) inndcommfail("connect");
+ r= ICCcommand('f',flushargv,&reply); if (r<0) inndcommfail("transmit");
+ if (!r) exit(0); /* yay! */
+
+ if (!strcmp(reply, "1 No such site")) exit(INNDCOMMCHILD_ESTATUS_NONESUCH);
+ syswarn("innd ctlinnd flush failed: innd said %s", reply);
+ exit(INNDCOMMCHILD_ESTATUS_FAIL);
+ }
+
+ close(pipefds[1]);
+ int sentinel_fd= pipefds[0];
+ on_fd_read_except(sentinel_fd, inndcomm_event);
+
+ SMS(FLUSHING, 0, why);
+}