fix assertion on unexpected vdp order for gzip-in-gunzip case

For the case of gzip included in plain response, the esi_level == 1 vdp
order was:

	pesi gunzip pesi_buf V2P(to_parent)

Yet we had assertions in place that pesi_buf always immediately follows
pesi.

The reason was that, for esi_level > 0, we would not push pesi_buf from
pesi init but rather from the transport, which was plain wrong: We
should delay any additional vdps in order to buffer the least amount of
data.

Working on this, I also noted that for the generic buffering case, our
assertion that pesi_buf is first, might be too strict. Now, any VDPs
before the buffer are being closed at esi_level > 1.

fixes this panic:

Assert error in vped_close_vdp(), vdp_pesi.c line 1182:
  Condition(vdpe->vdp == vdp) not true.

	...

Backtrace:
  0x43cf3b: /usr/sbin/varnishd() [0x43cf3b]
  0x4a01c2: /usr/sbin/varnishd(VAS_Fail+0x42) [0x4a01c2]
  0x7f3719306b63:
./vmod_cache/_vmod_pesi.4d9e0603bac2a2e2b2627f7fe90ff1d55d4759545517c869a5571f16636e230e(+0x8b63)
[0x7f3719306b63]
  0x7f371930a377:
./vmod_cache/_vmod_pesi.4d9e0603bac2a2e2b2627f7fe90ff1d55d4759545517c869a5571f16636e230e(+0xc377)
[0x7f371930a377]
  0x441eab: /usr/sbin/varnishd(CNT_Request+0x11ab) [0x441eab]
  0x7f3719308043:
./vmod_cache/_vmod_pesi.4d9e0603bac2a2e2b2627f7fe90ff1d55d4759545517c869a5571f16636e230e(+0xa043)
[0x7f3719308043]
  0x45c833: /usr/sbin/varnishd() [0x45c833]
  0x45ccf0: /usr/sbin/varnishd() [0x45ccf0]
  0x7f37e6d18e65: /lib64/libpthread.so.0(+0x7e65) [0x7f37e6d18e65]
  0x7f37e6a4188d: /lib64/libc.so.6(clone+0x6d) [0x7f37e6a4188d]
thread = (cache-worker)
pthread.attr = {
  guard = 4096,
  stack_bottom = 0x7f372c482000,
  stack_top = 0x7f372c502000,
  stack_size = 524288,
}
thr.req = 0x7f3601106020 {
  vxid = 25559391, transport = PESI_INCLUDE
  step = R_STP_TRANSMIT,
  req_body = R_BODY_NONE,
  restarts = 0, esi_level = 1,

	...

src/tests/gzip-in-plain.vtc

+varnishtest "gzipped level 1 in level 0 plain"
+
+server s1 {
+	rxreq
+	txresp -body {<Before esi1>
+<esi:include src="/esi1.html"/>
+Between esi1 and esi2
+<esi:include src="/esi2.html"/>
+After esi2
+}
+} -start
+
+server s2 {
+	rxreq
+	expect req.url == "/esi1.html"
+	txresp -gzipbody {<Before esi1-1>
+<esi:include src="/esi1-1.html"/>
+Between esi1-1 and esi1-2
+<esi:include src="/esi1-2.html"/>
+After esi1-2
+}
+} -start
+
+server s3 {
+	rxreq
+	expect req.url == "/esi2.html"
+	txresp -gzipbody {<Before esi2-1>
+<esi:include src="/esi2-1.html"/>
+Between esi2-1 and esi2-2
+<esi:include src="/esi2-2.html"/>
+After esi2-2
+}
+} -start
+
+server s4 {
+	loop 4 {
+		rxreq
+		expect req.url == "/esi1-1.html"
+		txresp -hdr "Cache-Control: max-age=0" -gzipbody {esi1-1
+}
+	}
+} -start
+
+server s5 {
+	loop 4 {
+		rxreq
+		expect req.url == "/esi1-2.html"
+		txresp -hdr "Cache-Control: max-age=0" -gzipbody {esi1-2
+}
+	}
+} -start
+
+server s6 {
+	loop 4 {
+		rxreq
+		expect req.url == "/esi2-1.html"
+		txresp -hdr "Cache-Control: max-age=0" -gzipbody {esi2-1
+}
+	}
+} -start
+
+server s7 {
+	loop 4 {
+		rxreq
+		expect req.url == "/esi2-2.html"
+		txresp -hdr "Cache-Control: max-age=0" -gzipbody {esi2-2
+}
+	}
+} -start
+
+varnish v1 -vcl+backend {
+	import ${vmod_pesi};
+
+	sub vcl_backend_fetch {
+		if (bereq.url == "/") {
+			set bereq.backend = s1;
+		}
+		elsif (bereq.url == "/esi1.html") {
+			set bereq.backend = s2;
+		}
+		elsif (bereq.url == "/esi2.html") {
+			set bereq.backend = s3;
+		}
+		elsif (bereq.url == "/esi1-1.html") {
+			set bereq.backend = s4;
+		}
+		elsif (bereq.url == "/esi1-2.html") {
+			set bereq.backend = s5;
+		}
+		elsif (bereq.url == "/esi2-1.html") {
+			set bereq.backend = s6;
+		}
+		elsif (bereq.url == "/esi2-2.html") {
+			set bereq.backend = s7;
+		}
+	}
+
+	sub vcl_backend_response {
+		set beresp.do_esi = true;
+	}
+
+	sub vcl_deliver {
+		pesi.activate();
+	}
+} -start
+
+client c1 {
+	txreq
+	rxresp
+	expect resp.status == 200
+	expect resp.bodylen == 191
+	expect resp.body == {<Before esi1>
+<Before esi1-1>
+esi1-1
+
+Between esi1-1 and esi1-2
+esi1-2
+
+After esi1-2
+
+Between esi1 and esi2
+<Before esi2-1>
+esi2-1
+
+Between esi2-1 and esi2-2
+esi2-2
+
+After esi2-2
+
+After esi2
+}
+} -run
+
+## HTTP/2
+
+varnish v1 -cliok "param.set feature +http2"
+
+## See comments in e23.vtc about limitations using vtc to test
+## gzipped H2 responses.
+
+client c1 {
+	stream 1 {
+		txreq
+		rxresp
+		expect resp.status == 200
+		expect resp.bodylen == 191
+		expect resp.body == {<Before esi1>
+<Before esi1-1>
+esi1-1
+
+Between esi1-1 and esi1-2
+esi1-2
+
+After esi1-2
+
+Between esi1 and esi2
+<Before esi2-1>
+esi2-1
+
+Between esi2-1 and esi2-2
+esi2-2
+
+After esi2-2
+
+After esi2
+}
+	} -run
+} -run

src/vdp_pesi.c

@@ -657,6 +657,7 @@ vdp_pesi_init(struct req *req, void **priv)
 #ifdef DEBUG_PESI_WS
 		pesi->ws_snap = WS_Snapshot(req->ws);
 #endif
+		VDP_Push(req, &VDP_pesi_buf, pesi);
 		return (0);
 	}
 
@@ -1193,6 +1194,19 @@ vped_close_vdp(struct req *req, int skip, const struct vdp *vdp)
 	vdc->nxt = nxt;
 }
 
+static const struct vdp *
+vped_next_vdp(struct req *req)
+{
+	struct vdp_entry *vdpe;
+	struct vdp_ctx *vdc;
+
+	vdc = req->vdc;
+	vdpe = VTAILQ_FIRST(&vdc->vdp);
+
+	CHECK_OBJ_NOTNULL(vdpe, VDP_ENTRY_MAGIC);
+	return (vdpe->vdp);
+}
+
 static void
 push_vdps(struct req *req, struct vped_gzgz_priv *vgzgz, struct nexus_gzip *gz)
 {
@@ -1220,6 +1234,7 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 	struct node *node, *parent;
 	struct nexus_gzip *gz;
 	struct vped_gzgz_priv *vgzgz = NULL;
+	const struct vdp *vdp;
 
 	VSLdbgv(req, "vped_deliver: req=%p boc=%p wantbody=%d", req, boc,
 		wantbody);
@@ -1293,8 +1308,7 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 	if (obj_esi) {
 		VSLdbg(req, "vped_deliver: ESI");
 
-		// VDP_pesi has been pushed from VCL
-		VDP_Push(req, &VDP_pesi_buf, pesi);
+		// VDP_pesi + buf has been pushed from VCL
 		push_vdps(req, vgzgz, gz);
 		AN(parent);
 		AZ(VDP_Push(req, &vped_to_parent, parent->nexus.req));
@@ -1323,6 +1337,11 @@ vped_deliver(struct req *req, struct boc *boc, int wantbody)
 		/* pesi_buf does not pesi_destroy() on fini */
 		req->transport_priv = NULL;
 		pesi_destroy(&pesi);
+		vdp = vped_next_vdp(req);
+		while (vdp != &VDP_pesi_buf) {
+			vped_close_vdp(req, 0, vdp);
+			vdp = vped_next_vdp(req);
+		}
 		vped_close_vdp(req, 0, &VDP_pesi_buf);
 		return;
 	}