imfile: improve truncation detection

previously, truncation was only detected at end of file. Especially with
busy files that could cause loss of data and possibly also stall imfile
reading. The new code now also checks during each read. Obviously, there
is some additional overhead associated with that, but this is unavoidable.

It still is highly recommended NOT to turn on "reopenOnTruncate" in imfile.
Note that there are also inherant reliability issues. There is no way to
"fix" these, as they are caused by races between the process(es) who truncate
and rsyslog reading the file. But with the new code, the "problem window"
should be much smaller and, more importantly, imfile should not stall.

see also https://github.com/rsyslog/rsyslog/issues/2659
see also https://github.com/rsyslog/rsyslog/issues/1605

runtime/rsyslog.h

@@ -221,9 +221,9 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	/* begin regular error codes */
 	RS_RET_NOT_IMPLEMENTED = -7,	/**< implementation is missing (probably internal error or lazyness ;)) */
 	RS_RET_OUT_OF_MEMORY = -6,	/**< memory allocation failed */
-	RS_RET_PROVIDED_BUFFER_TOO_SMALL = -50,
-/*< the caller provided a buffer, but the called function sees the size of this buffer is too small -
-operation not carried out */
+	RS_RET_PROVIDED_BUFFER_TOO_SMALL = -50, /*< the caller provided a buffer, but the called function sees
+						  the size of this buffer is too small - operation not carried out */
+	RS_RET_FILE_TRUNCATED = -51,	/**< (input) file was truncated, not an error but a status */
 	RS_RET_TRUE = -3,		/**< to indicate a true state (can be used as TRUE, legacy) */
 	RS_RET_FALSE = -2,		/**< to indicate a false state (can be used as FALSE, legacy) */
 	RS_RET_NO_IRET = -8,	/**< This is a trick for the debuging system - it means no iRet is provided  */

runtime/stream.c

@@ -400,6 +400,7 @@ static rsRetVal strmOpenFile(strm_t *pThis)
 	CHKiRet(doPhysOpen(pThis));
 
 	pThis->iCurrOffs = 0;
+	pThis->iBufPtrMax = 0;
 	CHKiRet(getFileSize(pThis->pszCurrFName, &offset));
 	if(pThis->tOperationsMode == STREAMMODE_WRITE_APPEND) {
 		pThis->iCurrOffs = offset;
@@ -636,6 +637,78 @@ finalize_it:
 	RETiRet;
 }
 
+
+/* helper to checkTruncation */
+static rsRetVal ATTR_NONNULL()
+rereadTruncated(strm_t *const pThis, const char *const reason)
+{
+	DEFiRet;
+
+	LogMsg(errno, RS_RET_FILE_TRUNCATED, LOG_WARNING, "file '%s': truncation detected, "
+		"(%s) - re-start reading from beginning",
+		pThis->pszCurrFName, reason);
+	DBGPRINTF("checkTruncation, file %s last buffer CHANGED\n", pThis->pszCurrFName);
+	CHKiRet(strmCloseFile(pThis));
+	CHKiRet(strmOpenFile(pThis));
+	iRet = RS_RET_FILE_TRUNCATED;
+
+finalize_it:
+	RETiRet;
+}
+/* helper to read:
+ * Check if file has been truncated since last read and, if so, re-set reading
+ * to begin of file. To detect truncation, we try to re-read the last block.
+ * If that does not succeed or different data than from the original read is
+ * returned, truncation is assumed.
+ * NOTE: this function must be called only if truncation is enabled AND
+ * when the previous read buffer still is valid (aka "before the next read").
+ * It is ok to call with a 0-size buffer, which we than assume as begin of
+ * reading. In that case, no truncation will be detected.
+ * rgerhards, 2018-09-20
+ */
+static rsRetVal ATTR_NONNULL()
+checkTruncation(strm_t *const pThis)
+{
+	DEFiRet;
+	int ret;
+	off64_t backseek;
+	assert(pThis->bReopenOnTruncate);
+
+	DBGPRINTF("checkTruncation, file %s, iBufPtrMax %zd\n", pThis->pszCurrFName, pThis->iBufPtrMax);
+	if(pThis->iBufPtrMax == 0) {
+		FINALIZE;
+	}
+
+	int currpos = lseek64(pThis->fd, 0, SEEK_CUR);
+	backseek = -1 * (off64_t) pThis->iBufPtrMax;
+	dbgprintf("checkTruncation in actual processing, currpos %d, backseek is %d\n", (int)currpos, (int) backseek);
+	ret = lseek64(pThis->fd, backseek, SEEK_CUR);
+	if(ret < 0) {
+		iRet = rereadTruncated(pThis, "cannot seek backward to begin of last block");
+		FINALIZE;
+	}
+dbgprintf("checkTruncation seek backwrds: %d\n", ret);
+currpos = lseek64(pThis->fd, 0, SEEK_CUR);
+dbgprintf("checkTruncation in actual processing, NEW currpos %d, backseek is %d\n", (int)currpos, (int) backseek);
+
+	const ssize_t lenRead = read(pThis->fd, pThis->pIOBuf_truncation, pThis->iBufPtrMax);
+	dbgprintf("checkTruncation proof-read: %d bytes\n", (int) lenRead);
+	if(lenRead < 0) {
+		iRet = rereadTruncated(pThis, "last block could not be re-read");
+		FINALIZE;
+	}
+
+	if(!memcmp(pThis->pIOBuf_truncation, pThis->pIOBuf, pThis->iBufPtrMax)) {
+		DBGPRINTF("checkTruncation, file %s last buffer unchanged\n", pThis->pszCurrFName);
+	} else {
+		iRet = rereadTruncated(pThis, "last block data different");
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
 /* read the next buffer from disk
  * rgerhards, 2008-02-13
  */
@@ -668,6 +741,13 @@ strmReadBuf(strm_t *pThis, int *padBytes)
 				toRead = (size_t) bytesLeft;
 			}
 		}
+		if(pThis->bReopenOnTruncate) {
+			rsRetVal localRet = checkTruncation(pThis);
+			if(localRet == RS_RET_FILE_TRUNCATED) {
+				continue;
+			}
+			CHKiRet(localRet);
+		}
 		iLenRead = read(pThis->fd, pThis->pIOBuf, toRead);
 		DBGOPRINT((obj_t*) pThis, "file %d read %ld bytes\n", pThis->fd, iLenRead);
 		/* end crypto */
@@ -1184,6 +1264,7 @@ static rsRetVal strmConstructFinalize(strm_t *pThis)
 	} else {
 		/* we work synchronously, so we need to alloc a fixed pIOBuf */
 		CHKmalloc(pThis->pIOBuf = (uchar*) MALLOC(pThis->sIOBufSize));
+		CHKmalloc(pThis->pIOBuf_truncation = (char*) MALLOC(pThis->sIOBufSize));
 	}
 
 finalize_it:
@@ -1231,6 +1312,7 @@ CODESTARTobjDestruct(strm)
 		}
 	} else {
 		free(pThis->pIOBuf);
+		free(pThis->pIOBuf_truncation);
 	}
 
 	/* Finally, we can free the resources.

runtime/stream.h

@@ -124,6 +124,7 @@ typedef struct strm_s {
 	ino_t inode;	/* current inode for files being monitored (undefined else) */
 	uchar *pszCurrFName; /* name of current file (if open) */
 	uchar *pIOBuf;	/* the iobuffer currently in use to gather data */
+	char *pIOBuf_truncation; /* iobuffer used during trucation detection block re-reads */
 	size_t iBufPtrMax;	/* current max Ptr in Buffer (if partial read!) */
 	size_t iBufPtr;	/* pointer into current buffer */
 	int iUngetC;	/* char set via UngetChar() call or -1 if none set */