ENT-14731: Added tests for cf-reactor

cf-agent/files_edit.h

@@ -41,6 +41,11 @@
                                    copy equal to the copied template file - not the
                                    copied + edited file. */
 
+/* The ifelapsed of the promises of edit bundles: CF_EDIT_IFELAPSED, unless the
+ * default ifelapsed is lower, e.g. 0 for the bundles cf-reactor runs on events,
+ * which may happen less than a minute apart */
+#define EDIT_IFELAPSED MIN(CF_EDIT_IFELAPSED, VIFELAPSED)
+
 typedef struct
 {
     char *filename;

cf-agent/files_editline.c

@@ -373,7 +373,7 @@ static PromiseResult VerifyLineDeletions(EvalContext *ctx, const Promise *pp, Ed
     char lockname[CF_BUFSIZE];
 
     Attributes a = GetDeletionAttributes(ctx, pp);
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     if (!SanityCheckDeletions(&a, pp))
     {
@@ -453,7 +453,7 @@ static PromiseResult VerifyColumnEdits(EvalContext *ctx, const Promise *pp, Edit
     char lockname[CF_BUFSIZE];
 
     Attributes a = GetColumnAttributes(ctx, pp);
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     if (a.column.column_separator == NULL)
     {
@@ -547,7 +547,7 @@ static PromiseResult VerifyPatterns(EvalContext *ctx, const Promise *pp, EditCon
 /* Are we working in a restricted region? */
 
     Attributes a = GetReplaceAttributes(ctx, pp);
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     if (!a.replace.replace_value)
     {
@@ -733,7 +733,7 @@ static PromiseResult VerifyLineInsertions(EvalContext *ctx, const Promise *pp, E
 
     Attributes a = GetInsertionAttributes(ctx, pp);
     int allow_multi_lines = StringEqual(a.sourcetype, "preserve_all_lines");
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     if (!SanityCheckInsertions(&a))
     {

cf-agent/files_editxml.c

@@ -353,7 +353,7 @@ static bool VerifyXPathBuild(EvalContext *ctx, const Attributes *attr, const Pro
     CfLock thislock;
     char lockname[CF_BUFSIZE], rawxpath[CF_BUFSIZE] = { 0 };
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     int ret = snprintf(rawxpath, sizeof(rawxpath), "%s",
         a.xml.havebuildxpath ? a.xml.build_xpath : pp->promiser);
@@ -417,7 +417,7 @@ static PromiseResult VerifyTreeDeletions(EvalContext *ctx, const Attributes *att
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     if (!SanityCheckTreeDeletions(&a))
     {
@@ -497,7 +497,7 @@ static PromiseResult VerifyTreeInsertions(EvalContext *ctx, const Attributes *at
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckTreeInsertions(&a, edcontext))
@@ -590,7 +590,7 @@ static PromiseResult VerifyAttributeDeletions(EvalContext *ctx, const Attributes
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckAttributeDeletions(&a))
@@ -673,7 +673,7 @@ static PromiseResult VerifyAttributeSet(EvalContext *ctx, const Attributes *attr
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckAttributeSet(&a))
@@ -756,7 +756,7 @@ static PromiseResult VerifyTextDeletions(EvalContext *ctx, const Attributes *att
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckTextDeletions(&a))
@@ -839,7 +839,7 @@ static PromiseResult VerifyTextSet(EvalContext *ctx, const Attributes *attr, con
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckTextSet(&a))
@@ -922,7 +922,7 @@ static PromiseResult VerifyTextInsertions(EvalContext *ctx, const Attributes *at
     CfLock thislock;
     char lockname[CF_BUFSIZE];
 
-    a.transaction.ifelapsed = CF_EDIT_IFELAPSED;
+    a.transaction.ifelapsed = EDIT_IFELAPSED;
 
     PromiseResult result = PROMISE_RESULT_NOOP;
     if (!SanityCheckTextInsertions(&a))

cf-reactor/reactor_transform.c

@@ -483,35 +483,36 @@ static PromiseResult RunThenBundles(EvalContext *ctx, const Promise *pp)
     return result;
 }
 
-typedef struct
-{
-    const char *promise_hash;
-    WatcherCheckResult check;
-} ReactorEventParam;
-
-/* Promise actuator for an events promise whose watcher fired, or failed to
- * check for its event. Only acts on the iteration the watcher was registered
- * for. */
-static PromiseResult KeepEventsPromiseOnEvent(EvalContext *ctx, const Promise *pp, void *param)
+/* Keeps the iteration of an events promise whose watcher fired, or failed to
+ * check for its event. */
+static PromiseResult KeepEventsPromiseOnEvent(EvalContext *ctx, const Promise *pp, WatcherCheckResult check)
 {
     assert(pp != NULL);
-    assert(param != NULL);
-    const ReactorEventParam *event = param;
 
-    char promise_hash[CF_HOSTKEY_STRING_SIZE];
-    WatcherPromiseHash(pp, promise_hash, sizeof(promise_hash));
-    if (!StringEqual(promise_hash, event->promise_hash))
+    PromiseResult result;
+    switch (check)
     {
-        return PROMISE_RESULT_SKIPPED;
+    case WATCHER_CHECK_EVENT:
+        result = RunThenBundles(ctx, pp);
+        break;
+
+    case WATCHER_CHECK_ERROR:
+        /* Already logged as an error by the watcher, see HandleQueuedEvents() */
+        result = PROMISE_RESULT_FAIL;
+        break;
+
+    default:
+        ProgrammingError("Unexpected watcher check result %d for reactor events promise '%s'",
+                         (int) check, pp->promiser);
     }
 
-    return RunThenBundles(ctx, pp);
+    RecordEventsPromiseOutcome(ctx, pp, result);
+    return result;
 }
 
-void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check)
+void HandleReactorEvent(EvalContext *ctx, const Promise *pp, WatcherCheckResult check)
 {
     assert(pp != NULL);
-    assert(promise_hash != NULL);
 
     /* cf-reactor keeps the same EvalContext across events. Like an agent run,
      * the bundle runs see the time classes of now and the persistent classes
@@ -523,11 +524,18 @@ void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise
     VariableTable *variables = NULL;
     EvalContextSnapshotTake(ctx, &classes, &variables);
 
-    ReactorEventParam event = { .promise_hash = promise_hash, .check = check };
-
+    /* pp is the iteration the watcher was registered with, expanded when its
+     * bundle was evaluated. It is kept as is rather than expanded again, as
+     * the bundle-local classes its class guard and if/unless were evaluated
+     * with are gone, and its functions could return something else. */
     EvalContextStackPushBundleFrame(ctx, PromiseGetBundle(pp), NULL, false, NULL);
     EvalContextStackPushBundleSectionFrame(ctx, pp->parent_section);
-    ExpandPromise(ctx, pp, KeepEventsPromiseOnEvent, &event);
+    EvalContextStackPushPromiseFrame(ctx, pp);
+    Promise *pexp = PromiseCopy(pp);
+    EvalContextStackPushExpandedPromiseIterationFrame(ctx, pexp);
+    KeepEventsPromiseOnEvent(ctx, pexp, check);
+    EvalContextStackPopFrame(ctx); /* promise iteration, destroys pexp */
+    EvalContextStackPopFrame(ctx); /* promise */
     EvalContextStackPopFrame(ctx); /* bundle section */
     EvalContextStackPopFrame(ctx); /* bundle */
 

cf-reactor/reactor_transform.h

@@ -47,13 +47,12 @@ void KeepReactorPromises(EvalContext *ctx, const Policy *policy);
  * outcome of the bundle as the outcome of the events promise. If the watcher
  * failed to check for its event, record a failure instead.
  *
- * @param pp the unexpanded events promise the watcher was registered for
- * @param promise_hash the hash of the expanded promise the watcher was
- *                     registered with, selecting the iteration of the events
- *                     promise to keep
+ * @param pp the expanded events promise the watcher was registered with,
+ *           kept as is: its class guard, if/unless and functions are not
+ *           evaluated again
  * @param check whether the event happened or the watcher failed to check for
  *              it (see WatcherEventFn)
  */
-void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check);
+void HandleReactorEvent(EvalContext *ctx, const Promise *pp, WatcherCheckResult check);
 
 #endif

cf-reactor/watcher.c

@@ -51,9 +51,8 @@ typedef struct
     WatcherCheckFn check_callback;
     WatcherStateDestroyFn destroy_state;
     void *state;
-    const Promise *promise; // the unexpanded events promise, owned by the policy
-    char *promise_hash; // hash of the expanded events promise, see WatcherPromiseHash()
-    char *promiser; // expanded promiser of the iteration of `promise`
+    Promise *promise; // copy of the expanded events promise, see PromiseCopy()
+    char *promise_hash; // hash of `promise`, see WatcherPromiseHash()
     time_t poll_interval_secs;
     time_t next_due;
     WatcherCheckResult last_check; // only touched by the watcher thread
@@ -174,8 +173,7 @@ bool WatcherRegister(const Promise *pp, WatcherOptions *opt)
 
     Watcher *w = xmalloc(sizeof(Watcher));
     w->promise_hash = xstrdup(promise_hash);
-    w->promise = pp->org_pp;
-    w->promiser = xstrdup(pp->promiser);
+    w->promise = PromiseCopy(pp);
     w->poll_interval_secs = opt->interval;
     w->next_due = 0; /* due immediately on the watcher thread's first pass */
     w->last_check = WATCHER_CHECK_NO_EVENT;
@@ -198,8 +196,8 @@ static void WatcherDestroy(void *item)
     {
         w->destroy_state(w->state);
     }
+    PromiseDestroy(w->promise);
     free(w->promise_hash);
-    free(w->promiser);
     free(w);
 }
 
@@ -299,15 +297,16 @@ bool EventWatcherInitialize(int *fd)
 
     *fd = WakeupChannelReadFd(&wakeup_channel);
 
-    Log(LOG_LEVEL_VERBOSE, "Started cf-reactor event watcher with %zu watcher(s)", MapSize(watchers));
+    /* No watchers yet, they are registered from the policy afterwards and
+     * counted by EventWatcherWakeUp() */
+    Log(LOG_LEVEL_VERBOSE, "Started cf-reactor event watcher");
     return true;
 }
 
 /* Watchers are only destroyed by the main thread (EventWatcherResume()),
  * which is also the one handling events, and never while events are queued,
  * so the watchers of queued events can be used without holding
- * watchers_mutex. Their promiser and promise are never modified by the
- * watcher thread. */
+ * watchers_mutex. Their promise is never modified by the watcher thread. */
 static void HandleQueuedEvents(EvalContext *ctx, WatcherEventFn on_event)
 {
     assert(on_event != NULL);
@@ -319,13 +318,13 @@ static void HandleQueuedEvents(EvalContext *ctx, WatcherEventFn on_event)
         const Watcher *w = event->watcher;
         if (event->check == WATCHER_CHECK_ERROR)
         {
-            Log(LOG_LEVEL_ERR, "Reactor watcher '%s' failed to check for its event", w->promiser);
+            Log(LOG_LEVEL_ERR, "Reactor watcher '%s' failed to check for its event", w->promise->promiser);
         }
         else
         {
-            Log(LOG_LEVEL_VERBOSE, "Reactor watcher '%s' fired", w->promiser);
+            Log(LOG_LEVEL_VERBOSE, "Reactor watcher '%s' fired", w->promise->promiser);
         }
-        on_event(ctx, w->promise, w->promise_hash, event->check);
+        on_event(ctx, w->promise, event->check);
         free(event);
     }
 }
@@ -395,6 +394,12 @@ void EventWatcherResume(void)
 
 void EventWatcherWakeUp(void)
 {
+    /* Called once the watchers of the (re-)read policy are registered */
+    ThreadLock(&watchers_mutex);
+    const size_t count = MapSize(watchers);
+    ThreadUnlock(&watchers_mutex);
+    Log(LOG_LEVEL_VERBOSE, "cf-reactor event watcher has %zu watcher(s)", count);
+
     if (watcher_thread != NULL)
     {
         StoppableThreadWake(watcher_thread);

cf-reactor/watcher.h

@@ -54,18 +54,16 @@ typedef struct
  * @brief Called on the main thread for every event of a registered watcher,
  *        and whenever a watcher starts failing to check for its event.
  *
- * @param pp the (unexpanded) events promise the watcher was registered for
- * @param promise_hash the hash of the expanded promise the watcher was
- *                     registered with (see WatcherPromiseHash()), selecting
- *                     the iteration of the events promise
+ * @param pp the expanded events promise the watcher was registered with (a
+ *           copy of it, owned by the watcher)
  * @param check WATCHER_CHECK_EVENT if the event happened, or
  *              WATCHER_CHECK_ERROR if the watcher failed to check for it
  *              (only reported once until a check succeeds again)
  * @note Called without the watcher registry locked, but it must not call
  *       EventWatcherPause(), as the events of the watchers are being
  *       handled.
  */
-typedef void (*WatcherEventFn)(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check);
+typedef void (*WatcherEventFn)(EvalContext *ctx, const Promise *pp, WatcherCheckResult check);
 
 /**
  * @brief Hash of an expanded events promise, identifying its watcher.
@@ -92,10 +90,10 @@ void WatcherRegistryFinalize(void);
  * reference for the next checks, so that only what happens from now on
  * fires. If that check fails, the watcher is not registered.
  *
- * @param pp the expanded events promise. Its hash and its unexpanded
- *           promise (pp->org_pp) are passed back to the WatcherEventFn on
- *           event. The unexpanded promise is not owned, so the policy
- *           holding it must not be destroyed before EventWatcherPause().
+ * @param pp the expanded events promise. The watcher keeps a copy of it
+ *           (see PromiseCopy()), passed back to the WatcherEventFn on event.
+ *           The copy refers to the policy holding pp, which must not be
+ *           destroyed before EventWatcherPause().
  * @param opt what the watcher checks for, see WatcherOptions. opt->state is
  *            owned by the registry, which destroys it right away if not
  *            used, so it must not be used by the caller afterwards.
@@ -132,7 +130,8 @@ void EventWatcherResume(void);
  * immediately. Without watchers, the watcher thread sleeps until woken up, so
  * this must be called after registering watchers outside of an
  * EventWatcherPause()/EventWatcherResume() pair (which wakes it up itself),
- * i.e. after the watchers of the initial policy.
+ * i.e. after the watchers of the initial policy. Logs the number of
+ * registered watchers.
  */
 void EventWatcherWakeUp(void);
 void EventWatcherFinalize(void);

libpromises/cf3parse_logic.h

@@ -230,11 +230,13 @@ static bool RelevantBundle(const char *agent, const char *blocktype)
         return true;
     }
 
-    /* Here are some additional bundle types handled by cfAgent */
+    /* Here are some additional bundle types handled by cfAgent, and by
+     * cf-reactor for the agent bundles it runs */
 
     Item *ip = SplitString("edit_line,edit_xml", ',');
 
-    if (strcmp(agent, CF_AGENTTYPES[AGENT_TYPE_AGENT]) == 0)
+    if (StringEqual(agent, CF_AGENTTYPES[AGENT_TYPE_AGENT]) ||
+        StringEqual(agent, CF_AGENTTYPES[AGENT_TYPE_REACTOR]))
     {
         if (IsItemIn(ip, blocktype))
         {

libpromises/eval_context.c

@@ -1746,6 +1746,16 @@ Promise *EvalContextStackPushPromiseIterationFrame(EvalContext *ctx, const Promi
     return pexp;
 }
 
+void EvalContextStackPushExpandedPromiseIterationFrame(EvalContext *ctx, Promise *pexp)
+{
+    assert(pexp != NULL);
+    assert(LastStackFrame(ctx, 0) != NULL);
+    assert(LastStackFrame(ctx, 0)->type == STACK_FRAME_TYPE_PROMISE);
+
+    EvalContextStackPushFrame(ctx, StackFrameNewPromiseIteration(pexp, NULL));
+    LoggingPrivSetLevels(CalculateLogLevel(pexp), CalculateReportLevel(pexp));
+}
+
 void EvalContextStackPopFrame(EvalContext *ctx)
 {
     assert(ctx != NULL);

libpromises/eval_context.h

@@ -218,6 +218,12 @@ void EvalContextStackPushBodyFrame(EvalContext *ctx, const Promise *caller, cons
 void EvalContextStackPushBundleSectionFrame(EvalContext *ctx, const BundleSection *owner);
 void EvalContextStackPushPromiseFrame(EvalContext *ctx, const Promise *owner);
 Promise *EvalContextStackPushPromiseIterationFrame(EvalContext *ctx, const PromiseIterator *iter_ctx);
+/**
+ * @brief Like EvalContextStackPushPromiseIterationFrame(), for an iteration
+ *        whose promise is already expanded: its constraints, class guard and
+ *        if/unless are not evaluated again. The frame takes ownership of pexp.
+ */
+void EvalContextStackPushExpandedPromiseIterationFrame(EvalContext *ctx, Promise *pexp);
 void EvalContextStackPopFrame(EvalContext *ctx);
 const char *EvalContextStackToString(EvalContext *ctx);
 void EvalContextSetBundleArgs(EvalContext *ctx, const Rlist *args);