fix valgrind errors

Author: sparrc

plugins/in_forward/fw_prot.c

@@ -1064,6 +1064,11 @@ static int fw_process_message_mode_entry(
 
     flb_log_event_encoder_reset(conn->ctx->log_encoder);
 
+    /* Check if connection was deleted during log append (e.g., due to pause) */
+    if (conn->being_deleted) {
+        return -1;
+    }
+
     if (chunk_id != -1) {
         chunk = options.via.map.ptr[chunk_id].val;
         send_ack(in, conn, chunk);
@@ -1468,6 +1473,14 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn)
                             out_tag, flb_sds_len(out_tag),
                             &entry.via.array.ptr[index],
                             chunk_id);
+
+                    /* Check if connection was deleted during processing */
+                    if (conn->being_deleted) {
+                        msgpack_unpacked_destroy(&result);
+                        msgpack_unpacker_free(unp);
+                        flb_sds_destroy(out_tag);
+                        return 0;
+                    }
                 }
 
                 if (chunk_id != -1) {
@@ -1641,6 +1654,16 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn)
 
                                     goto cleanup_decompress;
                                 }
+
+                                /* Check if connection was deleted during append */
+                                if (conn->being_deleted) {
+                                    flb_free(decomp_buf);
+                                    msgpack_unpacked_destroy(&result);
+                                    msgpack_unpacker_free(unp);
+                                    flb_sds_destroy(out_tag);
+                                    flb_decompression_context_destroy(conn->d_ctx);
+                                    return 0;
+                                }
                             }
                         } while (decomp_len > 0);