[6.0] b1db3bf45 vsl: Give up when it's too late to buffer records

[Martin Blix Grydeland]

commit b1db3bf45bef2594cf923a811918b4fe60805775
Author: Dridi Boukelmoune <dridi.boukelmoune at gmail.com>
Date:   Wed Oct 28 15:38:42 2020 +0100

    vsl: Give up when it's too late to buffer records
    
    When a VUT is slow enough, it might very well be overrun while it is
    scanning logs. For our built-in VUTs like varnishncsa or varnishlog
    this can happen if writing the output can block waiting for IO ops
    or when the output is piped to a slow consumer.

diff --git a/lib/libvarnishapi/vsl_dispatch.c b/lib/libvarnishapi/vsl_dispatch.c
index ebc45123a..2b4336b2a 100644
--- a/lib/libvarnishapi/vsl_dispatch.c
+++ b/lib/libvarnishapi/vsl_dispatch.c
@@ -427,7 +427,7 @@ chunk_shm_to_buf(struct VSLQ *vslq, struct chunk *chunk)
 }
 
 /* Append a set of records to a vtx structure */
-static void
+static enum vsl_status
 vtx_append(struct VSLQ *vslq, struct vtx *vtx, const struct VSLC_ptr *start,
     size_t len)
 {
@@ -439,6 +439,9 @@ vtx_append(struct VSLQ *vslq, struct vtx *vtx, const struct VSLC_ptr *start,
 	AN(start);
 
 	i = VSL_Check(vslq->c, start);
+	if (i == vsl_check_e_inval)
+		return (vsl_e_overrun);
+
 	if (i == vsl_check_valid && !VTAILQ_EMPTY(&vtx->shmchunks_free)) {
 		/* Shmref it */
 		chunk = VTAILQ_FIRST(&vtx->shmchunks_free);
@@ -453,7 +456,6 @@ vtx_append(struct VSLQ *vslq, struct vtx *vtx, const struct VSLC_ptr *start,
 		/* Append to shmref list */
 		VTAILQ_INSERT_TAIL(&vslq->shmrefs, chunk, shm.shmref);
 	} else {
-		assert(i != vsl_check_e_inval);
 		/* Buffer it */
 		chunk = VTAILQ_LAST(&vtx->chunks, chunkhead);
 		CHECK_OBJ_ORNULL(chunk, CHUNK_MAGIC);
@@ -468,6 +470,7 @@ vtx_append(struct VSLQ *vslq, struct vtx *vtx, const struct VSLC_ptr *start,
 		}
 	}
 	vtx->len += len;
+	return (vsl_more);
 }
 
 /* Allocate a new vtx structure */
@@ -1310,8 +1313,9 @@ vslq_next(struct VSLQ *vslq)
 		AN(vtx);
 	}
 	if (vtx != NULL) {
-		vtx_append(vslq, vtx, &c->rec, len);
-		vtx_scan(vslq, vtx);
+		r = vtx_append(vslq, vtx, &c->rec, len);
+		if (r == vsl_more)
+			vtx_scan(vslq, vtx);
 	}
 
 	return (r);