From: Christopher Faulet <cfaulet@haproxy.com>
Date: Thu, 25 Aug 2022 16:50:18 +0000 (+0200)
Subject: BUG/MEDIUM: spoe: Properly update streams waiting for a ACK in async mode
X-Git-Tag: v2.7-dev5~48
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=42a0662910816172469085cf739c1ff65871fdde;p=thirdparty%2Fhaproxy.git

BUG/MEDIUM: spoe: Properly update streams waiting for a ACK in async mode

A bug was introduced by the commit b042e4f6f ("BUG/MAJOR: spoe: properly
detach all agents when releasing the applet"). The fix is not correct. We
really want to known if the released appctx is the last one or not. It is
important when async mode is used. If there are still running applets, we
just need to remove the reference on the current applet from streams in the
async waiting queue.

With the commit above, in async mode, if there are still running applets, it
will work as expected. Otherwise a processing timeout will be reported for
all these streams. So it is not too bad. But for other modes (sync and
pipelining), the async waiting queue is always empty. If at least one stream
is waiting to send a message, a new applet is created. It is an issue if the
SPOA is unhealthy because the number of running applets may explode.

However, the commit above tried to fix an issue. The bug is in fact when an
new SPOE applet is created. On success, we must remove reference on the
current appctx from the streams in the async waiting queue.

This patch must be backported as far as 1.8.
---

diff --git a/src/flt_spoe.c b/src/flt_spoe.c
index 9973123f54..853452f533 100644
--- a/src/flt_spoe.c
+++ b/src/flt_spoe.c
@@ -1309,7 +1309,7 @@ spoe_release_appctx(struct appctx *appctx)
 	/* Destroy the task attached to this applet */
 	task_destroy(spoe_appctx->task);
 
-	/* Notify all waiting streams */
+	/* Report an error to all streams in the appctx waiting queue */
 	list_for_each_entry_safe(ctx, back, &spoe_appctx->waiting_queue, list) {
 		LIST_DELETE(&ctx->list);
 		LIST_INIT(&ctx->list);
@@ -1321,8 +1321,8 @@ spoe_release_appctx(struct appctx *appctx)
 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
 	}
 
-	/* If the applet was processing a fragmented frame, notify the
-	 * corresponding stream. */
+	/* If the applet was processing a fragmented frame, report an error to
+	 * the corresponding stream. */
 	if (spoe_appctx->frag_ctx.ctx) {
 		ctx = spoe_appctx->frag_ctx.ctx;
 		ctx->spoe_appctx = NULL;
@@ -1331,7 +1331,11 @@ spoe_release_appctx(struct appctx *appctx)
 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
 	}
 
-	if (!LIST_ISEMPTY(&agent->rt[tid].waiting_queue)) {
+	if (!LIST_ISEMPTY(&agent->rt[tid].applets)) {
+		/* If there are still some running applets, remove reference on
+		 * the current one from streams in the async waiting queue. In
+		 * async mode, the ACK may be received from another appctx.
+		 */
 		list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
 			if (ctx->spoe_appctx == spoe_appctx)
 				ctx->spoe_appctx = NULL;
@@ -1339,16 +1343,25 @@ spoe_release_appctx(struct appctx *appctx)
 		goto end;
 	}
 	else {
-		/* It is the last running applet and the sending and waiting
-		 * queues are not empty. Try to start a new one if HAproxy is
-		 * not stopping.
+		/* It is the last running applet and the sending and async
+		 * waiting queues are not empty. So try to start a new applet if
+		 * HAproxy is not stopping. On success, we remove reference on
+		 * the current appctx from streams in the async waiting queue.
+		 * In async mode, the ACK may be received from another appctx.
 		 */
 		if (!stopping &&
 		    (!LIST_ISEMPTY(&agent->rt[tid].sending_queue) || !LIST_ISEMPTY(&agent->rt[tid].waiting_queue)) &&
-		    spoe_create_appctx(agent->spoe_conf))
+		    spoe_create_appctx(agent->spoe_conf)) {
+			list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
+				if (ctx->spoe_appctx == spoe_appctx)
+					ctx->spoe_appctx = NULL;
+			}
 			goto end;
+		}
 
-		/* otherwise, notify all waiting streams */
+		/* Otherwise, report an error to all streams in the sending and
+		 * async waiting queues.
+		 */
 		list_for_each_entry_safe(ctx, back, &agent->rt[tid].sending_queue, list) {
 			LIST_DELETE(&ctx->list);
 			LIST_INIT(&ctx->list);