fix valgrind errors

Author: sparrc

plugins/in_forward/fw_prot.c

@@ -1016,7 +1016,10 @@ static int fw_process_message_mode_entry(
     msgpack_object   options;
     int              result;
     msgpack_object   chunk;
+    struct flb_in_fw_config *ctx;
 
+    /* Save ctx pointer before any operation that might delete the connection */
+    ctx = conn->ctx;
     metadata = NULL;
 
     if (chunk_id != -1 || metadata_id != -1) {
@@ -1030,39 +1033,47 @@ static int fw_process_message_mode_entry(
     result = flb_log_event_decoder_decode_timestamp(ts, &timestamp);
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
-        result = flb_log_event_encoder_begin_record(conn->ctx->log_encoder);
+        result = flb_log_event_encoder_begin_record(ctx->log_encoder);
     }
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
-        result = flb_log_event_encoder_set_timestamp(conn->ctx->log_encoder,
+        result = flb_log_event_encoder_set_timestamp(ctx->log_encoder,
                                                      &timestamp);
     }
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
         if (metadata != NULL) {
             result = flb_log_event_encoder_set_metadata_from_msgpack_object(
-                        conn->ctx->log_encoder,
+                        ctx->log_encoder,
                         metadata);
         }
     }
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
         result = flb_log_event_encoder_set_body_from_msgpack_object(
-                    conn->ctx->log_encoder,
+                    ctx->log_encoder,
                     body);
     }
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
-        result = flb_log_event_encoder_commit_record(conn->ctx->log_encoder);
+        result = flb_log_event_encoder_commit_record(ctx->log_encoder);
     }
 
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
         flb_input_log_append(in, tag, tag_len,
-                             conn->ctx->log_encoder->output_buffer,
-                             conn->ctx->log_encoder->output_length);
+                             ctx->log_encoder->output_buffer,
+                             ctx->log_encoder->output_length);
     }
 
-    flb_log_event_encoder_reset(conn->ctx->log_encoder);
+    flb_log_event_encoder_reset(ctx->log_encoder);
+
+    /* Check if plugin was paused during log append (connection may have been deleted) */
+    pthread_mutex_lock(&ctx->conn_mutex);
+    if (ctx->is_paused) {
+        pthread_mutex_unlock(&ctx->conn_mutex);
+        return -1;
+    }
+    pthread_mutex_unlock(&ctx->conn_mutex);
 
     if (chunk_id != -1) {
         chunk = options.via.map.ptr[chunk_id].val;
@@ -1468,6 +1479,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 +1660,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);