out_forward: close connection on failed flush to avoid frame desync

A forward frame is written as multiple non-atomic writes (header, body,
options). When one of them fails, part of the frame may already be on the
socket, so returning the connection to the keepalive pool lets the next
frame be read as the truncated frame's missing piece and desyncs the
receiver's msgpack stream. Force the connection closed on error so the
next flush starts on a fresh connection.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: zshuang0316 <zshuang0316@163.com>

Author: zshuang0316

plugins/out_forward/forward.c

@@ -1789,6 +1789,7 @@ static void cb_forward_flush(struct flb_event_chunk *event_chunk,
         flb_plg_debug(ctx->ins, "handshake status = %i", ret);
         if (ret == -1) {
             if (u_conn) {
+                flb_upstream_conn_recycle(u_conn, FLB_FALSE);
                 flb_upstream_conn_release(u_conn);
             }
 
@@ -1833,6 +1834,17 @@ static void cb_forward_flush(struct flb_event_chunk *event_chunk,
     }
 
     if (u_conn) {
+        if (ret != FLB_OK) {
+            /*
+             * A forward frame is written as multiple non-atomic writes
+             * (header, body, options). On failure part of it may already be
+             * on the socket, so don't return the connection to the keepalive
+             * pool: the next frame would be read as the truncated frame's
+             * missing piece and desync the receiver's msgpack stream. Force
+             * it closed so the next flush starts on a fresh connection.
+             */
+            flb_upstream_conn_recycle(u_conn, FLB_FALSE);
+        }
         flb_upstream_conn_release(u_conn);
     }