From 9782a805e0b9b39a0d59e95a9ed27ff42ee3e6f4 Mon Sep 17 00:00:00 2001
From: Balazs Scheidler <balazs.scheidler@axoflow.com>
Date: Sun, 5 May 2024 19:34:51 +0200
Subject: [PATCH] filterx: delegate per LogPipe filterx <> message sync to
 filterx

Signed-off-by: Balazs Scheidler <balazs.scheidler@axoflow.com>
---
 lib/filterx/filterx-eval.h | 22 ++++++++++++++++++++++
 lib/logmpx.c               | 13 ++++++-------
 lib/logpipe.h              |  9 ++-------
 3 files changed, 30 insertions(+), 14 deletions(-)

diff --git a/lib/filterx/filterx-eval.h b/lib/filterx/filterx-eval.h
index f58c9be1c64..f0d2a9f9c25 100644
--- a/lib/filterx/filterx-eval.h
+++ b/lib/filterx/filterx-eval.h
@@ -60,4 +60,26 @@ void filterx_eval_store_weak_ref(FilterXObject *object);
 void filterx_eval_init_context(FilterXEvalContext *context, FilterXEvalContext *previous_context);
 void filterx_eval_deinit_context(FilterXEvalContext *context);
 
+static inline void
+filterx_eval_sync_message(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  if (!context)
+    return;
+
+  if (!filterx_scope_is_dirty(context->scope))
+    return;
+
+  log_msg_make_writable(pmsg, path_options);
+  filterx_scope_sync(context->scope, *pmsg);
+}
+
+static inline void
+filterx_eval_prepare_for_fork(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  filterx_eval_sync_message(context, pmsg, path_options);
+  if (context)
+    filterx_scope_write_protect(context->scope);
+  log_msg_write_protect(*pmsg);
+}
+
 #endif
diff --git a/lib/logmpx.c b/lib/logmpx.c
index c129d0a73f3..eb47825351e 100644
--- a/lib/logmpx.c
+++ b/lib/logmpx.c
@@ -88,13 +88,12 @@ log_multiplexer_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_op
   log_path_options_push_junction(&local_path_options, &matched, path_options);
   if (_has_multiple_arcs(self))
     {
-      if (path_options->filterx_scope)
-        {
-          log_msg_make_writable(&msg, path_options);
-          filterx_scope_sync(path_options->filterx_scope, msg);
-          filterx_scope_write_protect(path_options->filterx_scope);
-        }
-      log_msg_write_protect(msg);
+      /* if we are delivering to multiple branches, we need to sync the
+       * filterx state with our message and also need to make everything
+       * write protected so that changes in those branches don't overwrite
+       * data we still need */
+
+      filterx_eval_prepare_for_fork(path_options->filterx_context, &msg, path_options);
     }
   for (fallback = 0; (fallback == 0) || (fallback == 1 && self->fallback_exists && !delivered); fallback++)
     {
diff --git a/lib/logpipe.h b/lib/logpipe.h
index fdf26a3b344..dc80c1c3bf1 100644
--- a/lib/logpipe.h
+++ b/lib/logpipe.h
@@ -458,13 +458,8 @@ log_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
         }
     }
 
-  if ((s->flags & PIF_SYNC_SCOPE) &&
-      path_options->filterx_scope &&
-      filterx_scope_is_dirty(path_options->filterx_scope))
-    {
-      log_msg_make_writable(&msg, path_options);
-      filterx_scope_sync(path_options->filterx_scope, msg);
-    }
+  if ((s->flags & PIF_SYNC_SCOPE))
+    filterx_eval_sync_message(path_options->filterx_context, &msg, path_options);
 
   if (G_UNLIKELY(s->flags & (PIF_HARD_FLOW_CONTROL | PIF_JUNCTION_END | PIF_CONDITIONAL_MIDPOINT)))
     {