ci: fix macOS shutdown, test fixtures, and browser teardown

.github/workflows/pr-closed-docker.yaml

@@ -12,7 +12,27 @@ jobs:
       # We may need a specific token here with `packages:admin` privileges which is not available to GITHUB_TOKEN
       packages: write
     steps:
+      - name: Check whether PR images exist
+        id: package
+        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        with:
+          script: |
+            try {
+              await github.rest.packages.getPackageForOrganization({
+                org: 'fluent',
+                package_type: 'container',
+                package_name: `fluent-bit/pr-${context.issue.number}`
+              });
+              return true;
+            } catch (error) {
+              if (error.status !== 404) {
+                throw error;
+              }
+              core.info('No container images were published for this PR.');
+              return false;
+            }
       - uses: vlaurin/action-ghcr-prune@0cf7d39f88546edd31965acba78cdcb0be14d641 # v0.6.0
+        if: steps.package.outputs.result == 'true'
         with:
           organization: fluent
           container: fluent-bit/pr-${{ github.event.number }}

plugins/out_s3/s3.c

@@ -57,16 +57,96 @@ struct worker_info {
 
 FLB_TLS_DEFINE(struct worker_info, s3_worker_info);
 
-#ifdef FLB_SYSTEM_WINDOWS
-static int setenv(const char *name, const char *value, int overwrite)
+struct s3_test_env_entry {
+    char *name;
+    char *value;
+    struct mk_list _head;
+};
+
+static pthread_mutex_t s3_test_env_mutex = PTHREAD_MUTEX_INITIALIZER;
+static struct mk_list s3_test_env = {&s3_test_env, &s3_test_env};
+
+/* Values remain immutable until reset, after all test workers have stopped. */
+char *flb_s3_test_env_get(const char *name)
 {
-    if (overwrite == 0 && getenv(name) != NULL) {
-        return 0;
+    struct mk_list *head;
+    struct s3_test_env_entry *entry;
+    char *value;
+
+    pthread_mutex_lock(&s3_test_env_mutex);
+    mk_list_foreach_r(head, &s3_test_env) {
+        entry = mk_list_entry(head, struct s3_test_env_entry, _head);
+        if (strcmp(entry->name, name) == 0) {
+            value = entry->value;
+            pthread_mutex_unlock(&s3_test_env_mutex);
+            return value;
+        }
+    }
+    pthread_mutex_unlock(&s3_test_env_mutex);
+    return getenv(name);
+}
+
+int flb_s3_test_env_set(const char *name, const char *value, int overwrite)
+{
+    struct s3_test_env_entry *entry;
+    struct s3_test_env_entry *previous;
+    struct mk_list *head;
+
+    entry = flb_calloc(1, sizeof(*entry));
+    if (entry == NULL) {
+        return -1;
+    }
+    entry->name = flb_strdup(name);
+    entry->value = value == NULL ? NULL : flb_strdup(value);
+    if (entry->name == NULL || (value != NULL && entry->value == NULL)) {
+        flb_free(entry->name);
+        flb_free(entry->value);
+        flb_free(entry);
+        return -1;
     }
 
-    return _putenv_s(name, value);
+    pthread_mutex_lock(&s3_test_env_mutex);
+    if (!overwrite) {
+        mk_list_foreach_r(head, &s3_test_env) {
+            previous = mk_list_entry(head, struct s3_test_env_entry, _head);
+            if (strcmp(previous->name, name) == 0) {
+                if (previous->value != NULL) {
+                    pthread_mutex_unlock(&s3_test_env_mutex);
+                    flb_free(entry->name);
+                    flb_free(entry->value);
+                    flb_free(entry);
+                    return 0;
+                }
+                break;
+            }
+        }
+    }
+    mk_list_add(&entry->_head, &s3_test_env);
+    pthread_mutex_unlock(&s3_test_env_mutex);
+    return 0;
+}
+
+int flb_s3_test_env_unset(const char *name)
+{
+    return flb_s3_test_env_set(name, NULL, 1);
+}
+
+void flb_s3_test_env_reset(void)
+{
+    struct mk_list *head;
+    struct mk_list *tmp;
+    struct s3_test_env_entry *entry;
+
+    pthread_mutex_lock(&s3_test_env_mutex);
+    mk_list_foreach_safe(head, tmp, &s3_test_env) {
+        entry = mk_list_entry(head, struct s3_test_env_entry, _head);
+        mk_list_del(&entry->_head);
+        flb_free(entry->name);
+        flb_free(entry->value);
+        flb_free(entry);
+    }
+    pthread_mutex_unlock(&s3_test_env_mutex);
 }
-#endif
 
 static int s3_timer_create(struct flb_s3 *ctx);
 
@@ -413,7 +493,7 @@ static char *mock_error_response(char *error_env_var)
     char *error = NULL;
     int len = 0;
 
-    err_val = getenv(error_env_var);
+    err_val = flb_s3_test_env_get(error_env_var);
     if (err_val != NULL && strlen(err_val) > 0) {
         error = flb_calloc(strlen(err_val) + 1, sizeof(char));
         if (error == NULL) {
@@ -432,7 +512,7 @@ static char *mock_error_response(char *error_env_var)
 
 int s3_plugin_under_test()
 {
-    if (getenv("FLB_S3_PLUGIN_UNDER_TEST") != NULL) {
+    if (flb_s3_test_env_get("FLB_S3_PLUGIN_UNDER_TEST") != NULL) {
         return FLB_TRUE;
     }
 
@@ -522,7 +602,7 @@ int create_headers(struct flb_s3 *ctx, char *body_md5,
 };
 
 /*
- * Track mock API call counts via env vars (e.g. TEST_PutObject_CALL_COUNT)
+ * Track mock API call counts in synchronized test state
  * so tests can assert the exact number of S3 API attempts.
  */
 static void mock_s3_call_increment_counter(char *api)
@@ -533,11 +613,11 @@ static void mock_s3_call_increment_counter(char *api)
     char buf[16];
 
     snprintf(env_var, sizeof(env_var), "TEST_%s_CALL_COUNT", api);
-    val = getenv(env_var);
+    val = flb_s3_test_env_get(env_var);
     count = val ? atoi(val) : 0;
     count++;
     snprintf(buf, sizeof(buf), "%d", count);
-    setenv(env_var, buf, 1);
+    flb_s3_test_env_set(env_var, buf, 1);
 }
 
 static void mock_s3_call_record_uri(char *api, char *uri)
@@ -546,18 +626,18 @@ static void mock_s3_call_record_uri(char *api, char *uri)
     char uri_env_var[64];
     char *count;
 
-    if (getenv("TEST_RECORD_S3_URIS") == NULL) {
+    if (flb_s3_test_env_get("TEST_RECORD_S3_URIS") == NULL) {
         return;
     }
 
     snprintf(count_env_var, sizeof(count_env_var), "TEST_%s_CALL_COUNT", api);
-    count = getenv(count_env_var);
+    count = flb_s3_test_env_get(count_env_var);
     if (count == NULL) {
         return;
     }
 
     snprintf(uri_env_var, sizeof(uri_env_var), "TEST_%s_URI_%s", api, count);
-    setenv(uri_env_var, uri, 1);
+    flb_s3_test_env_set(uri_env_var, uri, 1);
 }
 
 struct flb_http_client *mock_s3_call(char *error_env_var, char *api)
@@ -655,7 +735,7 @@ static struct flb_http_client *mock_s3_call_for_tag(char *error_env_var,
 {
     char *error_tag;
 
-    error_tag = getenv(error_tag_env_var);
+    error_tag = flb_s3_test_env_get(error_tag_env_var);
     if (error_tag != NULL &&
         (strlen(error_tag) != tag_len || memcmp(error_tag, tag, tag_len) != 0)) {
         return mock_s3_call("TEST_S3_OTHER_TAG_ERROR", api);
@@ -2100,7 +2180,7 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data,
     int ret;
 
     if (s3_plugin_under_test() == FLB_TRUE &&
-        getenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR") != NULL) {
+        flb_s3_test_env_get("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR") != NULL) {
         return -1;
     }
 

plugins/out_s3/s3.h

@@ -257,6 +257,10 @@ void multipart_upload_destroy(struct multipart_upload *m_upload);
 
 struct flb_http_client *mock_s3_call(char *error_env_var, char *api);
 int s3_plugin_under_test();
+char *flb_s3_test_env_get(const char *name);
+int flb_s3_test_env_set(const char *name, const char *value, int overwrite);
+int flb_s3_test_env_unset(const char *name);
+void flb_s3_test_env_reset(void);
 
 int get_md5_base64(char *buf, size_t buf_size, char *md5_str, size_t md5_str_size);
 

sdk/browser/fluent-bit-pthread.js

@@ -18,8 +18,9 @@ self.addEventListener('error', event => report('error', event.message || 'Pthrea
 self.addEventListener('message', event => {
     if (event.data?.flbShutdown === 1) {
         event.stopImmediatePropagation();
-        self.postMessage({flbPthread: 1, stage: 'closed'});
+        // Set the closing flag before allowing the owner to finish teardown.
         self.close();
+        self.postMessage({flbPthread: 1, stage: 'closed'});
         return;
     }
     // CMD_LOAD in the pinned Emscripten 6.0.9 libpthread.js.

sdk/browser/fluent-bit-worker.js

@@ -53,6 +53,7 @@ self.Worker = class extends BrowserWorker {
     #resolveReady;
     #rejectReady;
     #resolveClosed;
+    #shutdownPromise;
     #ready = new Promise((resolve, reject) => { this.#resolveReady = resolve; this.#rejectReady = reject; });
     constructor(url, options) {
         const prepared = preparedWorkers.get(options?.name);
@@ -104,19 +105,33 @@ self.Worker = class extends BrowserWorker {
         });
     }
     get ready() { return this.#ready; }
-    async shutdown() {
+    shutdown() {
+        if (this.#shutdownPromise) { return this.#shutdownPromise; }
         // Let initialized children close before their owner exits. Terminating
         // both levels at once can retain a Chromium worker after startup fails.
-        let timer;
-        try {
-            await new Promise(resolve => {
-                this.#resolveClosed = resolve;
-                timer = setTimeout(resolve, 250);
-                super.postMessage({flbShutdown: 1});
-            });
-        }
-        catch (_) { /* An unavailable child will be terminated below. */ }
-        finally { clearTimeout(timer); this.terminate(); }
+        this.#rejectReady(new Error('Pthread initialization cancelled'));
+        this.#load = null;
+        this.#shutdownPromise = (async () => {
+            let timer;
+            try {
+                await new Promise(resolve => {
+                    this.#resolveClosed = resolve;
+                    timer = setTimeout(resolve, 250);
+                    super.postMessage({flbShutdown: 1});
+                });
+            }
+            catch (_) { /* An unavailable child will be terminated below. */ }
+            finally {
+                clearTimeout(timer);
+                // Dispatch child termination while the owner is still alive.
+                // A self-close acknowledgement can precede Chromium releasing
+                // the nested WASM worker; let that termination finish first.
+                super.terminate();
+                await new Promise(resolve => setTimeout(resolve, 250));
+                pthreadWorkers.delete(this);
+            }
+        })();
+        return this.#shutdownPromise;
     }
     postMessage(...args) {
         if (args[0]?.cmd === PTHREAD_LOAD && !this.#runtimeReady) {
@@ -127,6 +142,12 @@ self.Worker = class extends BrowserWorker {
         super.postMessage(...args);
     }
     terminate() {
+        // Emscripten also terminates the pool during exit. Keep those children
+        // tracked until shutdown finishes so the SDK can await their cleanup.
+        if (this.#shutdownPromise || active?.operation === 4 || state === 'failed') {
+            void this.shutdown();
+            return;
+        }
         this.#rejectReady(new Error('Pthread initialization cancelled'));
         this.#load = null;
         pthreadWorkers.delete(this);
@@ -224,9 +245,9 @@ async function fatal(message, code = 'E_RUNTIME') {
     if (wasmInitialization) { await wasmInitialization.catch(() => {}); }
     releaseStorage?.();
     flushLogs();
-    post({type: 'fatal', code, message: String(message)});
     // Nothing can recover this runtime; release the worker's own event loop.
     self.close();
+    post({type: 'fatal', code, message: String(message), closed: true});
 }
 
 async function lockStorage(namespace) {
@@ -397,8 +418,8 @@ async function execute(request) {
     }
     if (method === 'destroy') {
         await command(id, 4);
-        await drainLogs();
         await Promise.all([...pthreadWorkers].map(worker => worker.shutdown()));
+        await drainLogs();
         releaseStorage?.();
         state = 'destroyed';
         return null;
@@ -432,11 +453,11 @@ self.onmessage = async ({data}) => {
     executing = true;
     try {
         const value = await execute(data);
-        post({type: 'response', id: data.id, ok: true, value});
         if (state === 'destroyed') {
             clearTimeout(logTimer);
             self.close();
         }
+        post({type: 'response', id: data.id, ok: true, value, closed: state === 'destroyed'});
     }
     catch (error) {
         await drainLogs();