[master] 729e69be0 vsl: Give up when it's too late to buffer records

[Dridi Boukelmoune]

commit 729e69be09f428906a57b22480f642c62c852ee2
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 cf11a92a0..a6d226c4e 100644
--- a/lib/libvarnishapi/vsl_dispatch.c
+++ b/lib/libvarnishapi/vsl_dispatch.c
@@ -431,7 +431,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)
 {
@@ -443,6 +443,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);
@@ -457,7 +460,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);
@@ -472,6 +474,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 */
@@ -1314,8 +1317,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);