out_s3: fix retry_limit semantics and multipart memory leaks

Fix retry_limit off-by-one: change all six >= comparisons to >
so retry_limit=N correctly allows N retries (N+1 total attempts),
matching engine semantics.

Default retry_limit to MAX_UPLOAD_ERRORS (5) when the user has
not explicitly set it, using the new retry_limit_is_set flag.
The engine default of 1 is too low for S3's internal retry
system where partially uploaded multipart data is wasted when
retries are exhausted too early. Log a warning when capping an
explicit unlimited setting to MAX_UPLOAD_ERRORS.

Fix multipart upload memory leaks:
- Add multipart_upload_destroy after mk_list_del when completion
  errors exceed retry_limit (80KB per abandoned upload)
- Free s3_file before flb_fstore_file_inactive when chunk
  failures exceed retry_limit (48 bytes per abandoned chunk)
- Free the json chunk buffer in all cb_s3_flush return paths

Fix use-after-free in s3_upload_queue retry-limit path: unlink
the multipart_upload from ctx->uploads before destroying it.
Without this, the freed node remains in the linked list and
get_upload() dereferences it on the next flush.

Fix double-free in s3_context_destroy: the upload_queue cleanup
loop was calling multipart_upload_destroy on uploads already
freed by the ctx->uploads cleanup loop above it.  Remove the
redundant destroy.

Fix mock_s3_call if/else chain so CreateMultipartUpload returns
a valid XML response with UploadId. Skip size validation in
test mode so multipart paths can be exercised with small data.
Remove unit_test_flush bypass so tests exercise the real flush
path. Add mock call counter and skip the timer floor in test
mode for faster execution.

Co-authored-by: Thean Lim <theanlim@amazon.com>
Signed-off-by: Anuj Singh <singholt@amazon.com>

Author: 2 people

plugins/out_s3/s3.c

@@ -259,6 +259,25 @@ int create_headers(struct flb_s3 *ctx, char *body_md5,
     return 0;
 };
 
+/*
+ * Track mock API call counts via env vars (e.g. TEST_PutObject_CALL_COUNT)
+ * so tests can assert the exact number of S3 API attempts.
+ */
+static void mock_s3_call_increment_counter(char *api)
+{
+    char env_var[64];
+    char *val;
+    int count;
+    char buf[16];
+
+    snprintf(env_var, sizeof(env_var), "TEST_%s_CALL_COUNT", api);
+    val = getenv(env_var);
+    count = val ? atoi(val) : 0;
+    count++;
+    snprintf(buf, sizeof(buf), "%d", count);
+    setenv(env_var, buf, 1);
+}
+
 struct flb_http_client *mock_s3_call(char *error_env_var, char *api)
 {
     /* create an http client so that we can set the response */
@@ -267,6 +286,8 @@ struct flb_http_client *mock_s3_call(char *error_env_var, char *api)
     char *resp;
     int len;
 
+    mock_s3_call_increment_counter(api);
+
     c = flb_calloc(1, sizeof(struct flb_http_client));
     if (!c) {
         flb_errno();
@@ -296,7 +317,7 @@ struct flb_http_client *mock_s3_call(char *error_env_var, char *api)
             "</InitiateMultipartUploadResult>";
             c->resp.payload_size = strlen(c->resp.payload);
         }
-        if (strcmp(api, "AbortMultipartUpload") == 0) {
+        else if (strcmp(api, "AbortMultipartUpload") == 0) {
             /* mocked success response */
             c->resp.status = 204;
             resp =            "Date:  Mon, 1 Nov 2010 20:34:56 GMT\n"
@@ -592,7 +613,6 @@ static void s3_context_destroy(struct flb_s3 *ctx)
     mk_list_foreach_safe(head, tmp, &ctx->upload_queue) {
         upload_contents = mk_list_entry(head, struct upload_queue, _head);
         s3_store_file_delete(ctx, upload_contents->upload_file);
-        multipart_upload_destroy(upload_contents->m_upload_file);
         remove_from_queue(upload_contents);
     }
 
@@ -630,7 +650,19 @@ static int cb_s3_init(struct flb_output_instance *ins,
     ctx->retry_time = 0;
     ctx->upload_queue_success = FLB_FALSE;
 
-    if(ctx->ins->retry_limit < 0) {
+    /*
+     * The engine default retry_limit (1) is too low for S3's internal
+     * retry system — partially uploaded multipart data is wasted when
+     * retries are exhausted too early. Default to MAX_UPLOAD_ERRORS (5)
+     * unless the user explicitly configured retry_limit.
+     */
+    if (ctx->ins->retry_limit_is_set == FLB_FALSE) {
+        ctx->ins->retry_limit = MAX_UPLOAD_ERRORS;
+    }
+    else if (ctx->ins->retry_limit < 0) {
+        flb_plg_warn(ctx->ins,
+                     "retry_limit set to unlimited, capping to %d",
+                     MAX_UPLOAD_ERRORS);
         ctx->ins->retry_limit = MAX_UPLOAD_ERRORS;
     }
 
@@ -734,12 +766,6 @@ static int cb_s3_init(struct flb_output_instance *ins,
     }
     flb_plg_info(ctx->ins, "Using upload size %lu bytes", ctx->file_size);
 
-    if (ctx->use_put_object == FLB_FALSE && ctx->file_size < 2 * MIN_CHUNKED_UPLOAD_SIZE) {
-            flb_plg_info(ctx->ins,
-                         "total_file_size is less than 10 MB, will use PutObject API");
-            ctx->use_put_object = FLB_TRUE;
-    }
-
     tmp = flb_output_get_property("compression", ins);
     if (tmp) {
         ret = flb_aws_compression_get_type(tmp);
@@ -761,6 +787,17 @@ static int cb_s3_init(struct flb_output_instance *ins,
     if (tmp) {
         ctx->content_type = (char *) tmp;
     }
+
+    if (s3_plugin_under_test() == FLB_TRUE) {
+        goto skip_size_validation;
+    }
+
+    if (ctx->use_put_object == FLB_FALSE && ctx->file_size < 2 * MIN_CHUNKED_UPLOAD_SIZE) {
+            flb_plg_info(ctx->ins,
+                         "total_file_size is less than 10 MB, will use PutObject API");
+            ctx->use_put_object = FLB_TRUE;
+    }
+    
     if (ctx->use_put_object == FLB_FALSE) {
         /* upload_chunk_size */
         if (ctx->upload_chunk_size <= 0) {
@@ -807,6 +844,7 @@ static int cb_s3_init(struct flb_output_instance *ins,
         }
     }
 
+skip_size_validation:
     tmp = flb_output_get_property("endpoint", ins);
     if (tmp) {
         ctx->insecure = strncmp(tmp, "http://", 7) == 0 ? FLB_TRUE : FLB_FALSE;
@@ -1016,11 +1054,13 @@ static int cb_s3_init(struct flb_output_instance *ins,
 
     ctx->timer_created = FLB_FALSE;
     ctx->timer_ms = (int) (ctx->upload_timeout / 6) * 1000;
-    if (ctx->timer_ms > UPLOAD_TIMER_MAX_WAIT) {
-        ctx->timer_ms = UPLOAD_TIMER_MAX_WAIT;
-    }
-    else if (ctx->timer_ms < UPLOAD_TIMER_MIN_WAIT) {
-        ctx->timer_ms = UPLOAD_TIMER_MIN_WAIT;
+    if (s3_plugin_under_test() == FLB_FALSE) {
+        if (ctx->timer_ms > UPLOAD_TIMER_MAX_WAIT) {
+            ctx->timer_ms = UPLOAD_TIMER_MAX_WAIT;
+        }
+        else if (ctx->timer_ms < UPLOAD_TIMER_MIN_WAIT) {
+            ctx->timer_ms = UPLOAD_TIMER_MIN_WAIT;
+        }
     }
 
     /*
@@ -1384,10 +1424,11 @@ static int put_all_chunks(struct flb_s3 *ctx)
                 continue;
             }
 
-            if (chunk->failures >= ctx->ins->retry_limit) {
+            if (chunk->failures > ctx->ins->retry_limit) {
                 flb_plg_warn(ctx->ins,
                              "Chunk for tag %s failed to send %d/%d times, will not retry",
                              (char *) fsf->meta_buf, chunk->failures, ctx->ins->retry_limit);
+                flb_free(chunk);
                 flb_fstore_file_inactive(ctx->fs, fsf);
                 continue;
             }
@@ -1672,7 +1713,7 @@ static struct multipart_upload *get_upload(struct flb_s3 *ctx,
         if (tmp_upload->upload_state == MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS) {
             continue;
         }
-        if (tmp_upload->upload_errors >= ctx->ins->retry_limit) {
+        if (tmp_upload->upload_errors > ctx->ins->retry_limit) {
             tmp_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS;
             flb_plg_error(ctx->ins, "Upload for %s has reached max upload errors",
                           tmp_upload->s3_key);
@@ -1918,10 +1959,13 @@ static void s3_upload_queue(struct flb_config *config, void *out_context)
 
             /* If retry limit was reached, discard file and remove file from queue */
             upload_contents->retry_counter++;
-            if (upload_contents->retry_counter >= ctx->ins->retry_limit) {
+            if (upload_contents->retry_counter > ctx->ins->retry_limit) {
                 flb_plg_warn(ctx->ins, "Chunk file failed to send %d times, will not "
                              "retry", upload_contents->retry_counter);
                 s3_store_file_inactive(ctx, upload_contents->upload_file);
+                if (upload_contents->m_upload_file) {
+                    mk_list_del(&upload_contents->m_upload_file->_head);
+                }
                 multipart_upload_destroy(upload_contents->m_upload_file);
                 remove_from_queue(upload_contents);
                 continue;
@@ -3319,10 +3363,11 @@ static void cb_s3_upload(struct flb_config *config, void *data)
         if (ret != FLB_OK) {
             flb_plg_error(ctx->ins, "Could not send chunk with tag %s",
                           (char *) fsf->meta_buf);
-            if(chunk->failures >= ctx->ins->retry_limit){
+            if(chunk->failures > ctx->ins->retry_limit){
                 flb_plg_warn(ctx->ins,
                              "Chunk for tag %s failed to send %d/%d times, will not retry",
                              (char *) fsf->meta_buf, chunk->failures, ctx->ins->retry_limit);
+                flb_free(chunk);
                 flb_fstore_file_inactive(ctx->fs, fsf);
                 continue;
             }
@@ -3334,11 +3379,12 @@ static void cb_s3_upload(struct flb_config *config, void *data)
         m_upload = mk_list_entry(head, struct multipart_upload, _head);
         complete = FLB_FALSE;
 
-        if (m_upload->complete_errors >= ctx->ins->retry_limit) {
+        if (m_upload->complete_errors > ctx->ins->retry_limit) {
             flb_plg_error(ctx->ins,
                           "Upload for %s has reached max completion errors, "
                           "plugin will give up", m_upload->s3_key);
             mk_list_del(&m_upload->_head);
+            multipart_upload_destroy(m_upload);
             continue;
         }
 
@@ -3528,31 +3574,6 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char
     return out_buf;
 }
 
-static void unit_test_flush(void *out_context, struct s3_file *upload_file,
-                            const char *tag, int tag_len, flb_sds_t chunk,
-                            int chunk_size, struct multipart_upload *m_upload_file,
-                            time_t file_first_log_time)
-{
-    int ret;
-    char *buffer;
-    size_t buffer_size;
-    struct flb_s3 *ctx = out_context;
-
-    s3_store_buffer_put(ctx, upload_file, tag, tag_len,
-                        chunk, (size_t) chunk_size, file_first_log_time);
-    ret = construct_request_buffer(ctx, chunk, upload_file, &buffer, &buffer_size);
-    if (ret < 0) {
-        flb_plg_error(ctx->ins, "Could not construct request buffer for %s",
-                      upload_file->file_path);
-        FLB_OUTPUT_RETURN(FLB_RETRY);
-    }
-
-    ret = upload_data(ctx, upload_file, m_upload_file, buffer, buffer_size, tag, tag_len);
-    flb_free(buffer);
-
-    FLB_OUTPUT_RETURN(ret);
-}
-
 static void flush_init(void *out_context)
 {
     int ret;
@@ -3838,16 +3859,8 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk,
         file_first_log_time = time(NULL);
     }
 
-    /* Specific to unit tests, will not get called normally */
-    if (s3_plugin_under_test() == FLB_TRUE) {
-        unit_test_flush(ctx, upload_file,
-                        event_chunk->tag, flb_sds_len(event_chunk->tag),
-                        chunk, chunk_size,
-                        m_upload_file, file_first_log_time);
-    }
-
     /* Discard upload_file if it has failed to upload retry_limit times */
-    if (upload_file != NULL && upload_file->failures >= ctx->ins->retry_limit) {
+    if (upload_file != NULL && upload_file->failures > ctx->ins->retry_limit) {
         flb_plg_warn(ctx->ins, "File with tag %s failed to send %d/%d times, will not retry",
                      event_chunk->tag, upload_file->failures, ctx->ins->retry_limit);
         s3_store_file_inactive(ctx, upload_file);