ENT-14195 1. Refactored cf-reactor event handling with one unified interface

cf-reactor/Makefile.am

@@ -38,7 +38,8 @@ AM_CFLAGS = $(CF3_CFLAGS) \
 libcf_reactor_la_LIBADD = ../libpromises/libpromises.la
 
 libcf_reactor_la_SOURCES = \
-	cf-reactor.c
+	cf-reactor.c cf-reactor.h \
+	reactor_context.c reactor_context.h
 
 if !BUILTIN_EXTENSIONS
  bin_PROGRAMS = cf-reactor

cf-reactor/cf-reactor.c

@@ -36,7 +36,7 @@
 #include <signal.h>             /* signal, kill */
 #include <signals.h>            /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon */
 #include <exec_tools.h>
-#include <alloc.h>              /* xmalloc */
+#include <reactor_context.h>
 
 /*****************************************************************************/
 /* Globals                                                                   */
@@ -190,38 +190,6 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv)
 /*****************************************************************************/
 
 
-static int SetupFileDescriptors(fd_set *readfds, int *fds, size_t num_fds)
-{
-    assert(readfds != NULL);
-
-    FD_ZERO(readfds);
-    int signal_pipe = GetSignalPipe();
-    FD_SET(signal_pipe, readfds);
-
-    int max_fd = signal_pipe;
-
-    for (size_t i = 0; i < num_fds; i++)
-    {
-        FD_SET(fds[i], readfds);
-        max_fd = MAX(fds[i], max_fd);
-    }
-    return max_fd + 1;
-}
-
-static bool ReactorNovaHasTimedOut(fd_set *readfds, int *fds, size_t num_fds)
-{
-    assert(readfds != NULL);
-
-    for (size_t i = 0; i < num_fds; i++)
-    {
-        if (FD_ISSET(fds[i], readfds))
-        {
-            return false;
-        }
-    }
-    return true;
-}
-
 int main(int argc, char *argv[])
 {
     GenericAgentConfig *config = CheckOpts(argc, argv);
@@ -268,28 +236,16 @@ int main(int argc, char *argv[])
     signal(SIGUSR1, HandleSignalsForDaemon);
     signal(SIGUSR2, HandleSignalsForDaemon);
 
-    /* Ask Nova how many fds it needs, rather than guessing a number here that
-     * really belongs to reactor-plugin (and would silently go stale if the
-     * two drift apart across releases). */
-    size_t max_nova_fds = ReactorNovaMaxFds();
-    int *all_fds = xmalloc(max_nova_fds * sizeof(int));
-    // the first num_nova_fds fds are populated with nova fds
-    size_t num_nova_fds;
-    if (!ReactorNovaInitialize(all_fds, max_nova_fds, &num_nova_fds))
+    ReactorContext reactor_ctx;
+    if (!ReactorContextInitialize(&reactor_ctx))
     {
-        free(all_fds);
         GenericAgentFinalize(ctx, config);
         DoCleanupAndExit(EXIT_FAILURE);
     }
-    // returns the number of fds used by nova reactor
-    size_t num_fds = num_nova_fds;
-    // TODO: populate all_fds with other fd used for event driven code (the
-    // allocation above will need to grow accordingly, e.g. by adding a fixed
-    // count on top of max_nova_fds before calling xmalloc())
 
     /* Writing to a pipe whose spawned process already exited (e.g. cfbs
      * rejecting its arguments before reading its stdin) must fail with EPIPE
-     * rather than terminate the whole daemon. Set after ReactorNovaInitialize(),
+     * rather than terminate the whole daemon. Set after ReactorContextInitialize(),
      * so that the spawner and the processes it execs keep the default handling. */
     signal(SIGPIPE, SIG_IGN);
 
@@ -298,22 +254,20 @@ int main(int argc, char *argv[])
     time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS;
     while (!IsPendingTermination())
     {
-        fd_set readfds;
-        int max_fd = SetupFileDescriptors(&readfds, all_fds, num_fds);
+        int max_fd = ReactorContextSetupFileDescriptors(&reactor_ctx);
 
         /* Determine how much time is remaining until the next tick. */
         time_t last_tick = time(NULL);
         time_t remaining = next_tick > last_tick ? next_tick - last_tick : 0;
 
         struct timeval timeout = { .tv_sec = remaining };
-        int ret = select(max_fd, &readfds, NULL, NULL, &timeout);
+        int ret = select(max_fd, &reactor_ctx.readfds, NULL, NULL, &timeout);
 
         /* Reschedule the backstop tick against the current time (not
          * `last_tick`, which was captured before select() potentially
          * blocked for the whole `remaining` duration), so that both call
-         * sites of ReactorNovaHandleTimeout() below agree on what "the next
-         * tick" means, instead of one of them silently doubling the
-         * interval. */
+         * sites of ReactorNovaHandleTimeout() agree on what "the next tick"
+         * means, instead of one of them silently doubling the interval. */
         next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS;
 
         if (ret < 0)
@@ -334,35 +288,14 @@ int main(int argc, char *argv[])
         else if (ret == 0)
         {
             /*** timeout ***/
-            Log(LOG_LEVEL_DEBUG, "Timed-out waiting for next notification");
-
             ReactorNovaHandleTimeout(&next_tick);
             continue;
         }
         /* else */
 
-        /* The signal pipe is always in the watched set so we wake up
-         * promptly on a pending signal, but (per its own contract in
-         * signals.c) it must be drained or it stays "ready" forever, which
-         * would stop select() from ever blocking again. */
-        if (FD_ISSET(GetSignalPipe(), &readfds))
-        {
-            unsigned char buf;
-            while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ }
-        }
-
-        /* This is needed since num_nova_fds may end up smaller than num_fds
-         * once other event-driven fds are added (see the TODO above). */
-        if (ReactorNovaHasTimedOut(&readfds, all_fds, num_nova_fds))
-        {
-            ReactorNovaHandleTimeout(&next_tick);
-            continue;
-        }
-
-        ReactorNovaHandleEvents(&readfds, all_fds, &next_tick);
+        ReactorContextHandleEvents(&reactor_ctx, &next_tick);
     }
-    ReactorNovaFinalize();
-    free(all_fds);
+    ReactorContextFinalize(&reactor_ctx);
 
     GenericAgentFinalize(ctx, config);
     CallCleanupFunctions();

cf-reactor/cf-reactor.h

@@ -0,0 +1,60 @@
+/*
+  Copyright 2026 Northern.tech AS
+
+  This file is part of CFEngine 3 - written and maintained by Northern.tech AS.
+
+  This program is free software; you can redistribute it and/or modify it
+  under the terms of the GNU General Public License as published by the
+  Free Software Foundation; version 3.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA
+
+  To the extent this program is licensed as part of the Enterprise
+  versions of CFEngine, the applicable Commercial Open Source License
+  (COSL) may apply to this file if you as a licensee so wish it. See
+  included file COSL.txt.
+*/
+
+#ifndef CFENGINE_REACTOR_H
+#define CFENGINE_REACTOR_H
+
+#include <platform.h>
+
+/**
+ * @brief Shared state for the cf-reactor daemon's single select(2) loop.
+ *
+ * `all_fds` is a single flat array shared by every event source the daemon
+ * watches. Nova's fds always occupy the first `num_nova_fds` slots (Nova
+ * owns that sub-range and is the only thing allowed to populate it); any
+ * other event source (currently just the watcher subsystem, see watcher.h)
+ * appends its own fd(s) after that, and `num_fds` tracks the total number
+ * of slots in use. Adding a new event source means:
+ *
+ *   1. Have it report how many fds it needs, and add that to the capacity
+ *      computed in ReactorContextInitialize() (reactor_context.c).
+ *   2. Give it an Initialize(fds, max_size, num_fds)/HandleEvents(readfds)/
+ *      Finalize(void) triplet shaped like ReactorNova*() or EventWatcher*(),
+ *      and wire the three calls into reactor_context.c next to the existing
+ *      ones.
+ *
+ * No other file needs to know how many event sources exist or in what
+ * order their fds appear.
+ */
+typedef struct ReactorContext
+{
+    int *all_fds;
+    size_t all_fds_capacity;
+    size_t num_nova_fds;
+    size_t num_fds;
+
+    fd_set readfds;
+} ReactorContext;
+
+#endif

cf-reactor/reactor_context.c

@@ -0,0 +1,120 @@
+/*
+  Copyright 2026 Northern.tech AS
+
+  This file is part of CFEngine 3 - written and maintained by Northern.tech AS.
+
+  This program is free software; you can redistribute it and/or modify it
+  under the terms of the GNU General Public License as published by the
+  Free Software Foundation; version 3.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA
+
+  To the extent this program is licensed as part of the Enterprise
+  versions of CFEngine, the applicable Commercial Open Source License
+  (COSL) may apply to this file if you as a licensee so wish it. See
+  included file COSL.txt.
+*/
+
+#include <reactor_context.h>
+#include <prototypes3.h>        /* ReactorNova*() */
+#include <signals.h>            /* GetSignalPipe() */
+#include <alloc.h>
+
+bool ReactorContextInitialize(ReactorContext *ctx)
+{
+    assert(ctx != NULL);
+
+    // TODO: initialize watcher registry
+
+    size_t max_nova_fds = ReactorNovaMaxFds();
+    ctx->all_fds_capacity = max_nova_fds + 1;
+    ctx->all_fds = xmalloc(ctx->all_fds_capacity * sizeof(int));
+
+    size_t num_nova_fds = 0;
+    if (!ReactorNovaInitialize(ctx->all_fds, max_nova_fds, &num_nova_fds))
+    {
+        WatcherRegistryFinalize();
+        free(ctx->all_fds);
+        ctx->all_fds = NULL;
+        return false;
+    }
+    ctx->num_nova_fds = num_nova_fds;
+    ctx->num_fds = num_nova_fds;
+
+    // TODO: intialize event watcher
+
+    return true;
+}
+
+int ReactorContextSetupFileDescriptors(ReactorContext *ctx)
+{
+    assert(ctx != NULL);
+
+    FD_ZERO(&ctx->readfds);
+    int signal_pipe = GetSignalPipe();
+    FD_SET(signal_pipe, &ctx->readfds);
+
+    int max_fd = signal_pipe;
+    for (size_t i = 0; i < ctx->num_fds; i++)
+    {
+        FD_SET(ctx->all_fds[i], &ctx->readfds);
+        max_fd = MAX(ctx->all_fds[i], max_fd);
+    }
+    return max_fd + 1;
+}
+
+static bool NovaHasTimedOut(ReactorContext *ctx)
+{
+    assert(ctx != NULL);
+    for (size_t i = 0; i < ctx->num_nova_fds; i++)
+    {
+        if (FD_ISSET(ctx->all_fds[i], &ctx->readfds))
+        {
+            return false;
+        }
+    }
+    return true;
+}
+
+void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick)
+{
+    assert(ctx != NULL);
+
+    if (NovaHasTimedOut(ctx))
+    {
+        ReactorNovaHandleTimeout(next_tick);
+    }
+    else
+    {
+        ReactorNovaHandleEvents(&ctx->readfds, ctx->all_fds, next_tick);
+    }
+
+    /* The signal pipe is always in the watched set so we wake up
+    * promptly on a pending signal, but (per its own contract in
+    * signals.c) it must be drained or it stays "ready" forever, which
+    * would stop select() from ever blocking again. */
+    if (FD_ISSET(GetSignalPipe(), &ctx->readfds))
+    {
+        unsigned char buf;
+        while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ }
+    }
+
+    // TODO: handle watcher events
+}
+
+void ReactorContextFinalize(ReactorContext *ctx)
+{
+    assert(ctx != NULL);
+
+    // TODO: finalize watcher events
+    ReactorNovaFinalize();
+    free(ctx->all_fds);
+    ctx->all_fds = NULL;
+}

cf-reactor/reactor_context.h

@@ -0,0 +1,36 @@
+/*
+  Copyright 2026 Northern.tech AS
+
+  This file is part of CFEngine 3 - written and maintained by Northern.tech AS.
+
+  This program is free software; you can redistribute it and/or modify it
+  under the terms of the GNU General Public License as published by the
+  Free Software Foundation; version 3.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA
+
+  To the extent this program is licensed as part of the Enterprise
+  versions of CFEngine, the applicable Commercial Open Source License
+  (COSL) may apply to this file if you as a licensee so wish it. See
+  included file COSL.txt.
+*/
+
+#ifndef CFENGINE_REACTOR_CONTEXT_H
+#define CFENGINE_REACTOR_CONTEXT_H
+
+#include <cf-reactor.h>
+
+
+bool ReactorContextInitialize(ReactorContext *ctx);
+int ReactorContextSetupFileDescriptors(ReactorContext *ctx);
+void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick);
+void ReactorContextFinalize(ReactorContext *ctx);
+
+#endif