From 9c10ad3fc9988072afd1125ddf392726fa51b558 Mon Sep 17 00:00:00 2001 From: Victor Moene Date: Thu, 8 Oct 2026 11:11:51 +0200 Subject: [PATCH 1/2] Made cf-reactor reuse an event state from a policy reload Instead of destroying all watchers, it now keeps the state, such that, even if an event happens during policy reload, it will get detected by comparing the previous state. Ticket: ENT-14702 Signed-off-by: Victor Moene --- cf-reactor/cf-reactor.c | 3 +- cf-reactor/file_watcher.c | 2 +- cf-reactor/file_watcher.h | 2 +- cf-reactor/reactor_transform.c | 42 +++----- cf-reactor/reactor_transform.h | 13 +-- cf-reactor/watcher.c | 177 ++++++++++++++++++--------------- cf-reactor/watcher.h | 82 +++++++++------ 7 files changed, 171 insertions(+), 150 deletions(-) diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index 4a93b9d64cc..405de519f02 100644 --- a/cf-reactor/cf-reactor.c +++ b/cf-reactor/cf-reactor.c @@ -41,7 +41,7 @@ #include #include #include -#include /* WatcherRegistryClear */ +#include /* EventWatcherPause, EventWatcherResume */ /*****************************************************************************/ /* Globals */ @@ -242,7 +242,6 @@ static void CheckPolicyUpdates(EvalContext *ctx, Policy **policy, GenericAgentCo * events detected so far with the current policy, and don't detect new * ones until the watchers of the new policy are registered */ EventWatcherPause(ctx, HandleReactorEvent); - WatcherRegistryClear(); EvalContextClear(ctx); PolicyDestroy(*policy); *policy = NULL; diff --git a/cf-reactor/file_watcher.c b/cf-reactor/file_watcher.c index b9b14ff72c4..8a82d06dff1 100644 --- a/cf-reactor/file_watcher.c +++ b/cf-reactor/file_watcher.c @@ -97,7 +97,7 @@ void *FileWatcherStateNew(const char *path) return (void *) fws; } -WatcherCheckResult CheckFileDeleted(void *state) +WatcherCheckResult FileWatcherCheckFileDeleted(void *state) { assert(state != NULL); FileWatcherState *fws = state; diff --git a/cf-reactor/file_watcher.h b/cf-reactor/file_watcher.h index ef1d47a762a..f96ea8daae5 100644 --- a/cf-reactor/file_watcher.h +++ b/cf-reactor/file_watcher.h @@ -30,7 +30,7 @@ void *FileWatcherStateNew(const char *path); -WatcherCheckResult CheckFileDeleted(void *state); +WatcherCheckResult FileWatcherCheckFileDeleted(void *state); void DestroyFileWatcherState(void *state); #endif diff --git a/cf-reactor/reactor_transform.c b/cf-reactor/reactor_transform.c index 7fa6bb86ad6..ad909c84994 100644 --- a/cf-reactor/reactor_transform.c +++ b/cf-reactor/reactor_transform.c @@ -139,11 +139,17 @@ static PromiseResult KeepEventsPromise(EvalContext *ctx, const Promise *pp) return PROMISE_RESULT_FAIL; } + WatcherOptions opt = { + .state = FileWatcherStateNew(path), + .check_callback = FileWatcherCheckFileDeleted, + .destroy_state = DestroyFileWatcherState, + }; + // register watcher Log(LOG_LEVEL_INFO, "Registering a file_deleted watcher for events promise '%s', on file '%s'", pp->promiser, path); - bool kept = WatcherRegister(pp->promiser, EVENT_FILE_DELETED, FileWatcherStateNew(path), pp->org_pp, 1); + WatcherRegister(pp, opt, 1); - return (kept) ? PROMISE_RESULT_NOOP : PROMISE_RESULT_FAIL; + return PROMISE_RESULT_NOOP; } static PromiseResult KeepReactorPromise(EvalContext *ctx, const Promise *pp, ARG_UNUSED void *param) @@ -250,7 +256,7 @@ static PromiseResult RunThenBundle(EvalContext *ctx, const Promise *pp) typedef struct { - const char *promiser; + const char *promise_hash; WatcherCheckResult check; } ReactorEventParam; @@ -263,36 +269,20 @@ static PromiseResult KeepEventsPromiseOnEvent(EvalContext *ctx, const Promise *p assert(param != NULL); const ReactorEventParam *event = param; - if (!StringEqual(pp->promiser, event->promiser)) + char promise_hash[CF_HOSTKEY_STRING_SIZE]; + WatcherPromiseHash(pp, promise_hash, sizeof(promise_hash)); + if (!StringEqual(promise_hash, event->promise_hash)) { return PROMISE_RESULT_SKIPPED; } - PromiseResult result; - switch (event->check) - { - case WATCHER_CHECK_EVENT: - result = RunThenBundle(ctx, pp); - break; - - case WATCHER_CHECK_ERROR: - Log(LOG_LEVEL_ERR, "Reactor events promise '%s' failed to check for its event", pp->promiser); - result = PROMISE_RESULT_FAIL; - break; - - default: - ProgrammingError("Unexpected watcher check result %d for reactor events promise '%s'", - (int) event->check, pp->promiser); - } - - RecordEventsPromiseOutcome(ctx, pp, result); - return result; + return RunThenBundle(ctx, pp); } -void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promiser, WatcherCheckResult check) +void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check) { assert(pp != NULL); - assert(promiser != 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 @@ -304,7 +294,7 @@ void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise VariableTable *variables = NULL; EvalContextSnapshotTake(ctx, &classes, &variables); - ReactorEventParam event = { .promiser = promiser, .check = check }; + ReactorEventParam event = { .promise_hash = promise_hash, .check = check }; EvalContextStackPushBundleFrame(ctx, PromiseGetBundle(pp), NULL, false, NULL); EvalContextStackPushBundleSectionFrame(ctx, pp->parent_section); diff --git a/cf-reactor/reactor_transform.h b/cf-reactor/reactor_transform.h index 6bb70fc0f42..01d2ff32bef 100644 --- a/cf-reactor/reactor_transform.h +++ b/cf-reactor/reactor_transform.h @@ -35,9 +35,9 @@ * (see watcher.h) for each of its "events" promises. * * Expects no watchers to be registered: on re-read of the policy, the - * watchers of the previous policy must be discarded with - * WatcherRegistryClear() before that policy is destroyed, as they refer to - * its promises. + * watchers of the previous policy must be set aside with + * EventWatcherPause() before that policy is destroyed, as they refer to its + * promises. */ void KeepReactorPromises(EvalContext *ctx, const Policy *policy); @@ -48,11 +48,12 @@ void KeepReactorPromises(EvalContext *ctx, const Policy *policy); * failed to check for its event, record a failure instead. * * @param pp the unexpanded events promise the watcher was registered for - * @param promiser the expanded promiser the watcher was registered with, - * selecting the iteration of the events promise to keep + * @param promise_hash the hash of the expanded promise the watcher was + * registered with, selecting the iteration of the events + * promise to keep * @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 *promiser, WatcherCheckResult check); +void HandleReactorEvent(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check); #endif diff --git a/cf-reactor/watcher.c b/cf-reactor/watcher.c index 8ca18680336..2fee123a574 100644 --- a/cf-reactor/watcher.c +++ b/cf-reactor/watcher.c @@ -29,11 +29,14 @@ #include // IsPendingTermination() #include #include -#include // StringEqual() -#include +#include // StringEqual_untyped(), StringHash_untyped() +#include #include #include #include // ThreadLock(), ThreadUnlock() +#include // PromiseRuntimeHash() +#include // HashPrintSafe() +#include // Promise /* Upper bound on how long the watcher thread ever sleeps in one go, so that * IsPendingTermination() is re-checked at least this often during shutdown, @@ -44,11 +47,12 @@ typedef struct { - char *promiser; // expanded promiser of the iteration of `promise` - WatcherCheckFn check_callback; // resolved from `type` at WatcherRegister() time + 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` time_t poll_interval_secs; time_t next_due; WatcherCheckResult last_check; // only touched by the watcher thread @@ -63,12 +67,13 @@ typedef struct static void WatcherDestroy(void *item); -/* Guards `watchers` and `paused` against the watcher thread +/* Guards `watchers`, `old_watchers` and `paused` against the watcher thread * (WatcherThreadMain()) iterating over `watchers` concurrently with the main * thread re-registering watchers from a freshly (re-)read policy - * (WatcherRegister(), WatcherRegistryClear()). */ + * (WatcherRegister(), EventWatcherPause()). */ static pthread_mutex_t watchers_mutex = PTHREAD_MUTEX_INITIALIZER; -static Seq *watchers = NULL; +static Map *watchers = NULL; +static Map *old_watchers = NULL; static bool paused = false; /* see EventWatcherPause() */ static WakeupChannel wakeup_channel = { .fds = { -1, -1 } }; @@ -76,119 +81,104 @@ static WakeupChannel wakeup_channel = { .fds = { -1, -1 } }; /* Events of watchers (WatcherEvent), queued by the watcher thread for the * main thread to handle. The watchers are not owned: the watcher thread only * queues events of watchers of the registry while holding watchers_mutex, and - * WatcherRegistryClear() requires the queue to be empty, so the watchers of + * EventWatcherPause() empties the queue before setting the watchers aside, so the watchers of * the queued events are always registered. */ static ThreadedQueue *event_queue = NULL; static StoppableThread *watcher_thread = NULL; /*****************************************************************************/ -void WatcherRegistryInitialize(void) +/* The hash covers the whole expanded promise, including what is watched + * (the constraints of the 'when' body) and what is run ('then'), so that a + * state is only taken over by a watcher of the same type, on the same + * thing. */ +void WatcherPromiseHash(const Promise *pp, char *hash, size_t hash_size) { - assert(watchers == NULL); + assert(pp != NULL); + assert(hash != NULL); - watchers = SeqNew(4, WatcherDestroy); + unsigned char digest[EVP_MAX_MD_SIZE + 1]; + PromiseRuntimeHash(pp, "events_promise_hash", digest, CF_DEFAULT_DIGEST); + HashPrintSafe(hash, hash_size, digest, CF_DEFAULT_DIGEST, true); } -void WatcherRegistryFinalize(void) -{ - SeqDestroy(watchers); - watchers = NULL; -} - -void WatcherRegistryClear(void) +void WatcherRegistryInitialize(void) { - assert(watchers != NULL); - - ThreadLock(&watchers_mutex); - - if (event_queue != NULL && !ThreadedQueueIsEmpty(event_queue)) - { - ProgrammingError("Clearing the reactor watcher registry with events of its " - "watchers still queued, call EventWatcherPause() first"); - } - - SeqDestroy(watchers); - watchers = SeqNew(4, WatcherDestroy); + assert(watchers == NULL); - ThreadUnlock(&watchers_mutex); + watchers = MapNew(StringHash_untyped, StringEqual_untyped, NULL, WatcherDestroy); } -/* A watcher is identified by its events promise and the expanded promiser - * of the iteration: the iterations of a promise share the same (unexpanded) - * promise, and different promises can have the same promiser. */ -static const Watcher *FindWatcher(const Promise *pp, const char *promiser) +void WatcherRegistryFinalize(void) { - for (size_t i = 0; i < SeqLength(watchers); i++) - { - const Watcher *w = SeqAt(watchers, i); - if (w->promise == pp && StringEqual(w->promiser, promiser)) - { - return w; - } - } - return NULL; + MapDestroy(watchers); + watchers = NULL; + MapDestroy(old_watchers); + old_watchers = NULL; } // Expects interval to be strictly greater than 0, otherwise the watcher thread will busy spin -bool WatcherRegister(const char *promiser, EventType type, void *state, const Promise *pp, time_t interval) +void WatcherRegister(const Promise *pp, WatcherOptions opt, time_t interval) { - assert(promiser != NULL); assert(pp != NULL); + assert(pp->promiser != NULL); + assert(opt.check_callback != NULL); + assert(opt.destroy_state != NULL); + assert(opt.state != NULL); assert(watchers != NULL); assert(interval > 0); - - WatcherCheckFn check_callback = NULL; - WatcherStateDestroyFn destroy_state = NULL; - switch (type) - { - case EVENT_FILE_DELETED: - check_callback = CheckFileDeleted; - destroy_state = DestroyFileWatcherState; - break; - - // TODO: add more event types - - default: - ProgrammingError("Unknown reactor event type %d for watcher '%s'", (int) type, promiser); - } - + ThreadLock(&watchers_mutex); + char promise_hash[CF_HOSTKEY_STRING_SIZE]; + WatcherPromiseHash(pp, promise_hash, sizeof(promise_hash)); - if (FindWatcher(pp, promiser) != NULL) + Watcher *prev = MapGet(watchers, promise_hash); + if (prev != NULL) { - Log(LOG_LEVEL_ERR, "Reactor watcher '%s' is already registered, ignoring the duplicate", promiser); + // ignore duplicates ThreadUnlock(&watchers_mutex); - if (destroy_state != NULL) - { - destroy_state(state); - } - return false; + opt.destroy_state(opt.state); + return; } Watcher *w = xmalloc(sizeof(Watcher)); - w->promiser = xstrdup(promiser); - w->state = state; - w->promise = pp; + w->promise_hash = xstrdup(promise_hash); + w->promise = pp->org_pp; + w->promiser = xstrdup(pp->promiser); w->poll_interval_secs = interval; w->next_due = 0; /* due immediately on the watcher thread's first pass */ w->last_check = WATCHER_CHECK_NO_EVENT; - w->check_callback = check_callback; - w->destroy_state = destroy_state; + w->check_callback = opt.check_callback; + w->destroy_state = opt.destroy_state; + + // Take over the state of an identical watcher + prev = (old_watchers != NULL) ? MapGet(old_watchers, promise_hash) : NULL; + if (prev == NULL) + { + w->state = opt.state; + } + else + { + Log(LOG_LEVEL_DEBUG, "Reactor watcher '%s' takes over the state of an identical watcher", w->promiser); + w->state = prev->state; + prev->state = NULL; /* so not destroyed with prev */ + opt.destroy_state(opt.state); + } - SeqAppend(watchers, w); + MapInsert(watchers, w->promise_hash, w); ThreadUnlock(&watchers_mutex); - return true; } static void WatcherDestroy(void *item) { Watcher *w = item; - if (w->destroy_state != NULL) + // w->state is NULL when it was borrowed by a new watcher + if (w->destroy_state != NULL && w->state != NULL) { w->destroy_state(w->state); } + free(w->promise_hash); free(w->promiser); free(w); } @@ -210,9 +200,11 @@ static void WatcherThreadMain(StoppableThread *thread, ARG_UNUSED void *unused) ThreadLock(&watchers_mutex); - for (size_t i = 0; !paused && i < SeqLength(watchers); i++) + MapIterator it = MapIteratorInit(watchers); + MapKeyValue *item; + while (!paused && (item = MapIteratorNext(&it)) != NULL) { - Watcher *w = SeqAt(watchers, i); + Watcher *w = item->value; // check_callback cannot be null if it is correctly assigned during policy evaluation assert(w->check_callback != NULL); @@ -276,11 +268,11 @@ bool EventWatcherInitialize(int *fd) *fd = WakeupChannelReadFd(&wakeup_channel); - Log(LOG_LEVEL_VERBOSE, "Started reactor watcher subsystem with %zu watcher(s)", SeqLength(watchers)); + Log(LOG_LEVEL_VERBOSE, "Started cf-reactor event watcher with %zu watcher(s)", MapSize(watchers)); return true; } -/* Watchers are only destroyed by the main thread (WatcherRegistryClear()), +/* 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 @@ -302,7 +294,7 @@ static void HandleQueuedEvents(EvalContext *ctx, WatcherEventFn on_event) { Log(LOG_LEVEL_NOTICE, "Reactor watcher '%s' fired", w->promiser); } - on_event(ctx, w->promise, w->promiser, event->check); + on_event(ctx, w->promise, w->promise_hash, event->check); free(event); } } @@ -338,11 +330,32 @@ void EventWatcherPause(EvalContext *ctx, WatcherEventFn on_event) { HandleQueuedEvents(ctx, on_event); } + + ThreadLock(&watchers_mutex); + + /* Paused, so the watcher thread can't have queued any event since */ + if (event_queue != NULL && !ThreadedQueueIsEmpty(event_queue)) + { + ProgrammingError("Events of reactor watchers still queued after handling them while pausing"); + } + + /* Set the watchers aside before the policy is re-read, for the watchers + * registered from it to take over the state of the identical ones, see + * WatcherRegister(). If paused twice in a row, the watchers of the last + * policy are the ones registered since the first pause. */ + MapDestroy(old_watchers); + old_watchers = watchers; + watchers = MapNew(StringHash_untyped, StringEqual_untyped, NULL, WatcherDestroy); + + ThreadUnlock(&watchers_mutex); } void EventWatcherResume(void) { ThreadLock(&watchers_mutex); + /* The watchers of the previous policy not taken over */ + MapDestroy(old_watchers); + old_watchers = NULL; paused = false; ThreadUnlock(&watchers_mutex); } diff --git a/cf-reactor/watcher.h b/cf-reactor/watcher.h index 3c2726afd50..58c66a62ba5 100644 --- a/cf-reactor/watcher.h +++ b/cf-reactor/watcher.h @@ -26,11 +26,7 @@ #define CFENGINE_WATCHER_H #include /* Bundle */ - -typedef enum -{ - EVENT_FILE_DELETED, -} EventType; +#include /* CF_HOSTKEY_STRING_SIZE */ typedef enum { @@ -42,63 +38,85 @@ typedef enum typedef WatcherCheckResult (*WatcherCheckFn)(void *state); typedef void (*WatcherStateDestroyFn)(void *state); +/** + * @brief What a watcher checks for, and how, depending on the type of event. + */ +typedef struct +{ + WatcherCheckFn check_callback; /* whether the event happened, updating state */ + WatcherStateDestroyFn destroy_state; + void *state; /* owned by the registry once registered */ +} WatcherOptions; + /** * @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 promiser the expanded promiser the watcher was registered with, - * selecting the iteration of the events promise + * @param promise_hash the hash of the expanded promise the watcher was + * registered with (see WatcherPromiseHash()), selecting + * the iteration of the events promise * @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 - * WatcherRegistryClear(), as the events of the watchers are being + * EventWatcherPause(), as the events of the watchers are being * handled. */ -typedef void (*WatcherEventFn)(EvalContext *ctx, const Promise *pp, const char *promiser, WatcherCheckResult check); - -void WatcherRegistryInitialize(void); -void WatcherRegistryFinalize(void); +typedef void (*WatcherEventFn)(EvalContext *ctx, const Promise *pp, const char *promise_hash, WatcherCheckResult check); /** - * @brief Discard every currently registered watcher and start over. Call - * this before re-registering watchers from a freshly (re-)read policy. + * @brief Hash of an expanded events promise, identifying its watcher. * - * @note No events of the watchers may be queued, i.e. once the event watcher - * is initialized, call EventWatcherPause() first. + * @param pp the expanded events promise + * @param hash where to write the hash, of size CF_HOSTKEY_STRING_SIZE */ -void WatcherRegistryClear(void); +void WatcherPromiseHash(const Promise *pp, char *hash, size_t hash_size); + +void WatcherRegistryInitialize(void); +void WatcherRegistryFinalize(void); /** * @brief Register a specific watcher instance. * - * @param promiser the expanded promiser of the iteration of the events - * promise. Together with pp, identifies the watcher. - * @param type the type of watcher, defined in when bodies - * @param state the data used for by the watcher, depending on the type - * @param pp the unexpanded events promise, passed back to the WatcherEventFn - * on event. Not owned, so the registry must be cleared before the - * policy holding it is destroyed. + * Watchers are identified by the hash of their expanded promise (see + * WatcherPromiseHash()). A watcher identical to a registered one is + * ignored, so that the 'then' bundle isn't run twice for the same event. + * + * If a watcher of the previous policy (see EventWatcherPause()) was + * registered for an identical expanded promise, the new watcher takes over + * its state instead of opt.state, so that the events happening while paused + * are detected. + * + * @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 opt what the watcher checks for, see WatcherOptions. opt.state is + * owned by the registry. * @param interval interval between runs */ -bool WatcherRegister(const char *promiser, EventType type, void *state, const Promise *pp, time_t interval); +void WatcherRegister(const Promise *pp, WatcherOptions opt, time_t interval); bool EventWatcherInitialize(int *fd); void EventWatcherHandleEvents(EvalContext *ctx, WatcherEventFn on_event, int fd, fd_set *readfds); /** - * @brief Stop checking for events, and handle the events already queued, so - * that the watchers can be cleared (see WatcherRegistryClear()) without - * losing any event. Events are checked for again after EventWatcherResume(). + * @brief Stop checking for events, handle the events already queued, and + * set the registered watchers aside, so that the policy they refer to can be + * destroyed and the watchers of the new policy registered (see + * WatcherRegister()). Events are checked for again after + * EventWatcherResume(). * - * @note Only the events already detected are guaranteed to be handled: - * whether an event happening while paused is detected afterwards - * depends on the state the watcher is registered again with. + * @note An event happening while paused is detected after + * EventWatcherResume() if a watcher for an identical promise is + * registered again, see WatcherRegister(). */ void EventWatcherPause(EvalContext *ctx, WatcherEventFn on_event); /** - * @brief Check for events again, after EventWatcherPause(). + * @brief Check for events again, after EventWatcherPause(). The watchers set + * aside by EventWatcherPause() are destroyed, and the registered ones are + * checked immediately. */ void EventWatcherResume(void); void EventWatcherFinalize(void); From e10206ac0c9a3921d4d1e5936857af7b1fb5cade Mon Sep 17 00:00:00 2001 From: Victor Moene Date: Thu, 8 Oct 2026 12:55:04 +0200 Subject: [PATCH 2/2] Updated watcher_test.c Signed-off-by: Victor Moene --- tests/unit/watcher_test.c | 386 ++++++++++++++++++++++++++++---------- 1 file changed, 284 insertions(+), 102 deletions(-) diff --git a/tests/unit/watcher_test.c b/tests/unit/watcher_test.c index 007a0fcacc0..365ce620c62 100644 --- a/tests/unit/watcher_test.c +++ b/tests/unit/watcher_test.c @@ -1,7 +1,7 @@ #include #include -#include /* FileWatcherStateNew() */ +#include /* FileWatcherStateNew(), FileWatcherCheckFileDeleted() */ #include /* Promise */ #include /* LoggingPrivContext, LoggingPrivSetContext() */ @@ -24,16 +24,91 @@ static char *CaptureErrorLogHook(ARG_UNUSED LoggingPrivContext *pctx, LogLevel l return (char *) message; } -/* The registry never dereferences the promise it is given, it only hands it - * back to the WatcherEventFn, so a dummy one is enough. */ -static const Promise dummy_promise; +/* The registry only hashes the expanded promise it is given (its promiser + * here, as it has no constraints, comment or parent) and hands the + * unexpanded one (org_pp) back to the WatcherEventFn, so minimal promises + * are enough. */ +static const Promise unexpanded_promise; +static const Promise expanded_promise = { + .promiser = "test-event", + .org_pp = &unexpanded_promise, +}; + +static WatcherOptions FileDeletedOptions(const char *path) +{ + WatcherOptions opt = { + .state = FileWatcherStateNew(path), + .check_callback = FileWatcherCheckFileDeleted, + .destroy_state = DestroyFileWatcherState, + }; + return opt; +} + +/* Creates the file 'file' in a new temporary directory dir */ +static void CreateTempFile(char *dir, char *path, size_t path_size) +{ + assert_true(mkdtemp(dir) != NULL); + snprintf(path, path_size, "%s/file", dir); + int file_fd = open(path, O_CREAT | O_WRONLY, 0600); + assert_true(file_fd >= 0); + close(file_fd); +} + +/* Waits up to tenths/10 seconds for the watcher thread to queue an event, + * without handling it */ +static bool WaitForQueuedEvent(int fd, int tenths) +{ + for (int i = 0; i < tenths; i++) + { + fd_set readfds; + FD_ZERO(&readfds); + FD_SET(fd, &readfds); + struct timeval timeout = { .tv_sec = 0, .tv_usec = 100000 }; + if (select(fd + 1, &readfds, NULL, NULL, &timeout) > 0) + { + return true; + } + } + return false; +} static int handled_event_count = 0; +static const Promise *last_event_promise = NULL; +static char last_event_promise_hash[CF_HOSTKEY_STRING_SIZE]; +static WatcherCheckResult last_event_check = WATCHER_CHECK_NO_EVENT; -static void CountEvent(ARG_UNUSED EvalContext *ctx, ARG_UNUSED const Promise *pp, - ARG_UNUSED const char *promiser, ARG_UNUSED WatcherCheckResult check) +static void RecordEvent(ARG_UNUSED EvalContext *ctx, const Promise *pp, + const char *promise_hash, WatcherCheckResult check) { handled_event_count++; + last_event_promise = pp; + strlcpy(last_event_promise_hash, promise_hash, sizeof(last_event_promise_hash)); + last_event_check = check; +} + +/* Asserts that the last event was handled for the iteration `pp` */ +static void AssertLastEventFor(const Promise *pp) +{ + char promise_hash[CF_HOSTKEY_STRING_SIZE]; + WatcherPromiseHash(pp, promise_hash, sizeof(promise_hash)); + assert_true(last_event_promise == pp->org_pp); + assert_string_equal(last_event_promise_hash, promise_hash); +} + +static void ResetRecordedEvents(void) +{ + handled_event_count = 0; + last_event_promise = NULL; + last_event_promise_hash[0] = '\0'; + last_event_check = WATCHER_CHECK_NO_EVENT; +} + +static void HandleEvents(int fd) +{ + fd_set readfds; + FD_ZERO(&readfds); + FD_SET(fd, &readfds); + EventWatcherHandleEvents(NULL, RecordEvent, fd, &readfds); } static void test_registry_initialize_finalize(void) @@ -46,56 +121,85 @@ static void test_watcher_register_single(void) { WatcherRegistryInitialize(); - assert_true(WatcherRegister("test-event", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/test-event"), &dummy_promise, 5)); + WatcherRegister(&expanded_promise, FileDeletedOptions("/nonexistent/test-event"), 5); WatcherRegistryFinalize(); } -static void test_watcher_register_duplicate_ignored(void) +static void test_watcher_register_iterations(void) { WatcherRegistryInitialize(); - assert_true(WatcherRegister("dup-event", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/dup-event-a"), &dummy_promise, 5)); + /* The iterations of a promise (same promise, other promisers) are + * different watchers */ + const Promise iteration_a = { .promiser = "/a", .org_pp = &unexpanded_promise }; + const Promise iteration_b = { .promiser = "/b", .org_pp = &unexpanded_promise }; + WatcherRegister(&iteration_a, FileDeletedOptions("/nonexistent/a"), 5); + WatcherRegister(&iteration_b, FileDeletedOptions("/nonexistent/b"), 5); - captured_err_count = 0; - captured_err_message[0] = '\0'; - /* force_hook_level makes the hook still see the error while the console - * level is lowered, so the expected error isn't printed. */ - LoggingPrivContext log_ctx = { - .log_hook = CaptureErrorLogHook, - .force_hook_level = LOG_LEVEL_ERR, + WatcherRegistryFinalize(); +} + +static void test_watcher_register_identical_ignored(void) +{ + char dir[] = "/tmp/watcher_test.XXXXXX"; + char path[PATH_MAX]; + CreateTempFile(dir, path, sizeof(path)); + + WatcherRegistryInitialize(); + + /* Another promise identical to a registered one (same expanded promise, + * so same hash) is ignored, so that the 'then' bundle isn't run twice + * for the same event. The state of the ignored one is destroyed by the + * registry (checked by valgrind/ASan). */ + static const Promise other_unexpanded_promise; + const Promise other_expanded_promise = { + .promiser = "test-event", + .org_pp = &other_unexpanded_promise, }; - LoggingPrivSetContext(&log_ctx); - const LogLevel old_level = LogGetGlobalLevel(); - LogSetGlobalLevel(LOG_LEVEL_CRIT); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); + WatcherRegister(&other_expanded_promise, FileDeletedOptions(path), 1); - /* Registering the same promise and promiser again must be rejected: an error is logged - * (not silently swallowed) and the first registration is kept, not - * replaced. */ - const bool registered = WatcherRegister("dup-event", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/dup-event-b"), &dummy_promise, 5); + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); - LogSetGlobalLevel(old_level); - LoggingPrivSetContext(NULL); + ResetRecordedEvents(); + assert_int_equal(unlink(path), 0); - assert_false(registered); - assert_int_equal(captured_err_count, 1); - assert_true(strstr(captured_err_message, "dup-event") != NULL); - assert_true(strstr(captured_err_message, "already registered") != NULL); + assert_true(WaitForQueuedEvent(fd, 50)); + HandleEvents(fd); - WatcherRegistryFinalize(); + /* Handled once, with the promise registered first */ + assert_int_equal(handled_event_count, 1); + AssertLastEventFor(&expanded_promise); + + EventWatcherFinalize(); + assert_int_equal(rmdir(dir), 0); } -static void test_watcher_register_same_promiser_of_other_promise(void) +static void test_watcher_register_again_after_pause(void) { WatcherRegistryInitialize(); - /* Watchers are identified by their promise and promiser: iterations of - * a promise (same promise, other promisers), and promises with the same - * promiser (other promises), are all different watchers */ - static const Promise other_promise; - assert_true(WatcherRegister("/a", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/a"), &dummy_promise, 5)); - assert_true(WatcherRegister("/b", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/b"), &dummy_promise, 5)); - assert_true(WatcherRegister("/a", EVENT_FILE_DELETED, FileWatcherStateNew("/nonexistent/a"), &other_promise, 5)); + /* Pausing sets the registered watchers aside, so the same promise can be + * registered again for the new policy, also without the event watcher + * being initialized */ + ResetRecordedEvents(); + WatcherRegister(&expanded_promise, FileDeletedOptions("/nonexistent/test-event"), 5); + EventWatcherPause(NULL, RecordEvent); + WatcherRegister(&expanded_promise, FileDeletedOptions("/nonexistent/test-event"), 5); + EventWatcherResume(); + + /* Paused twice in a row (e.g. a failed policy reload in between), the + * watchers set aside the first time are replaced by those registered + * since, without leaking (checked by valgrind/ASan) */ + EventWatcherPause(NULL, RecordEvent); + WatcherRegister(&expanded_promise, FileDeletedOptions("/nonexistent/test-event"), 5); + EventWatcherPause(NULL, RecordEvent); + WatcherRegister(&expanded_promise, FileDeletedOptions("/nonexistent/test-event"), 5); + EventWatcherResume(); + + assert_int_equal(handled_event_count, 0); WatcherRegistryFinalize(); } @@ -110,16 +214,16 @@ static void test_event_watcher_lifecycle(void) fd_set readfds; FD_ZERO(&readfds); - handled_event_count = 0; + ResetRecordedEvents(); /* fd not set: must return early without touching the wakeup channel or * the event queue. */ - EventWatcherHandleEvents(NULL, CountEvent, fd, &readfds); + EventWatcherHandleEvents(NULL, RecordEvent, fd, &readfds); FD_SET(fd, &readfds); /* fd set but nothing queued: must drain the channel (no-op) and find * nothing to pop from the event queue. */ - EventWatcherHandleEvents(NULL, CountEvent, fd, &readfds); + EventWatcherHandleEvents(NULL, RecordEvent, fd, &readfds); assert_int_equal(handled_event_count, 0); /* Must wake up and stop the watcher thread by itself (IsPendingTermination() @@ -127,77 +231,151 @@ static void test_event_watcher_lifecycle(void) EventWatcherFinalize(); } -/* Waits up to timeout_secs for the watcher thread to signal a queued event - * on fd, without handling it */ -static bool WaitForQueuedEvent(int fd, int timeout_secs) +static void test_event_watcher_pause_handles_queued_events(void) { - for (int i = 0; i < timeout_secs * 10; i++) - { - fd_set readfds; - FD_ZERO(&readfds); - FD_SET(fd, &readfds); - struct timeval timeout = { .tv_sec = 0, .tv_usec = 100000 }; - if (select(fd + 1, &readfds, NULL, NULL, &timeout) > 0) - { - return true; - } - } - return false; + char dir[] = "/tmp/watcher_test.XXXXXX"; + char path[PATH_MAX]; + CreateTempFile(dir, path, sizeof(path)); + + WatcherRegistryInitialize(); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); + + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); + assert_int_equal(unlink(path), 0); + + /* Wait for the watcher thread to queue the event (it checks every + * second), without handling it */ + assert_true(WaitForQueuedEvent(fd, 50)); + + /* Pausing handles the queued event, with the unexpanded promise and the + * promise hash of its watcher, so that the watchers can be set aside without + * losing it */ + ResetRecordedEvents(); + EventWatcherPause(NULL, RecordEvent); + assert_int_equal(handled_event_count, 1); + AssertLastEventFor(&expanded_promise); + assert_int_equal(last_event_check, WATCHER_CHECK_EVENT); + + EventWatcherResume(); + + EventWatcherFinalize(); + assert_int_equal(rmdir(dir), 0); } -static void HandleEvents(int fd, WatcherEventFn on_event) +static void test_event_watcher_state_kept_across_pause(void) { - fd_set readfds; - FD_ZERO(&readfds); - FD_SET(fd, &readfds); - EventWatcherHandleEvents(NULL, on_event, fd, &readfds); -} + char dir[] = "/tmp/watcher_test.XXXXXX"; + char path[PATH_MAX]; + CreateTempFile(dir, path, sizeof(path)); -static const Promise *last_event_promise = NULL; + WatcherRegistryInitialize(); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); -static WatcherCheckResult last_event_check = WATCHER_CHECK_NO_EVENT; + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); -static void RecordEvent(ARG_UNUSED EvalContext *ctx, const Promise *pp, - ARG_UNUSED const char *promiser, WatcherCheckResult check) -{ - handled_event_count++; - last_event_promise = pp; - last_event_check = check; + ResetRecordedEvents(); + EventWatcherPause(NULL, RecordEvent); + assert_int_equal(handled_event_count, 0); + + /* Deleted while the policy is reloaded: a new state would never see the + * file, so only the state kept from the previous watcher detects it */ + assert_int_equal(unlink(path), 0); + + /* The identical promise of the new policy, i.e. a new unexpanded + * promise, but the same expanded one */ + static const Promise new_unexpanded_promise; + const Promise new_expanded_promise = { + .promiser = "test-event", + .org_pp = &new_unexpanded_promise, + }; + WatcherRegister(&new_expanded_promise, FileDeletedOptions(path), 1); + EventWatcherResume(); + + assert_true(WaitForQueuedEvent(fd, 50)); + HandleEvents(fd); + + /* Handled with the promise of the new policy, not the destroyed one */ + assert_int_equal(handled_event_count, 1); + AssertLastEventFor(&new_expanded_promise); + + EventWatcherFinalize(); + assert_int_equal(rmdir(dir), 0); } -static void test_event_watcher_pause_handles_queued_events(void) +static void test_event_watcher_state_kept_identical_ignored(void) { char dir[] = "/tmp/watcher_test.XXXXXX"; - assert_true(mkdtemp(dir) != NULL); char path[PATH_MAX]; - snprintf(path, sizeof(path), "%s/file", dir); - int file_fd = open(path, O_CREAT | O_WRONLY, 0600); - assert_true(file_fd >= 0); - close(file_fd); + CreateTempFile(dir, path, sizeof(path)); WatcherRegistryInitialize(); - assert_true(WatcherRegister("pause-event", EVENT_FILE_DELETED, FileWatcherStateNew(path), &dummy_promise, 1)); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); int fd = -1; assert_true(EventWatcherInitialize(&fd)); + + ResetRecordedEvents(); + EventWatcherPause(NULL, RecordEvent); assert_int_equal(unlink(path), 0); - /* Wait for the watcher thread to queue the event (it checks every - * second), without handling it */ - assert_true(WaitForQueuedEvent(fd, 5)); + /* Two identical promises in the new policy: the first takes over the + * state of the previous watcher, so the deletion is still detected, and + * the second is ignored */ + static const Promise new_unexpanded_promise; + static const Promise other_unexpanded_promise; + const Promise new_expanded_promise = { + .promiser = "test-event", + .org_pp = &new_unexpanded_promise, + }; + const Promise other_expanded_promise = { + .promiser = "test-event", + .org_pp = &other_unexpanded_promise, + }; + WatcherRegister(&new_expanded_promise, FileDeletedOptions(path), 1); + WatcherRegister(&other_expanded_promise, FileDeletedOptions(path), 1); + EventWatcherResume(); + + assert_true(WaitForQueuedEvent(fd, 50)); + HandleEvents(fd); - /* Pausing handles the queued event, with the promise of its watcher, so - * that the registry can be cleared without losing it */ - handled_event_count = 0; - last_event_promise = NULL; - EventWatcherPause(NULL, RecordEvent); assert_int_equal(handled_event_count, 1); - assert_true(last_event_promise == &dummy_promise); - assert_int_equal(last_event_check, WATCHER_CHECK_EVENT); + AssertLastEventFor(&new_expanded_promise); + + EventWatcherFinalize(); + assert_int_equal(rmdir(dir), 0); +} + +static void test_event_watcher_state_not_kept_for_other_promise(void) +{ + char dir[] = "/tmp/watcher_test.XXXXXX"; + char path[PATH_MAX]; + CreateTempFile(dir, path, sizeof(path)); + + WatcherRegistryInitialize(); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); - WatcherRegistryClear(); + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); + + ResetRecordedEvents(); + EventWatcherPause(NULL, RecordEvent); + assert_int_equal(unlink(path), 0); + + /* A different promise in the new policy, even on the same file, starts + * with a new state, which never saw the file */ + const Promise other_expanded_promise = { + .promiser = "other-event", + .org_pp = &unexpanded_promise, + }; + WatcherRegister(&other_expanded_promise, FileDeletedOptions(path), 1); EventWatcherResume(); + /* The watcher thread checks at least every second */ + assert_false(WaitForQueuedEvent(fd, 30)); + assert_int_equal(handled_event_count, 0); + EventWatcherFinalize(); assert_int_equal(rmdir(dir), 0); } @@ -230,7 +408,7 @@ static void test_check_file_deleted_error(void) LogSetGlobalLevel(LOG_LEVEL_CRIT); void *state = FileWatcherStateNew(path); - assert_int_equal(CheckFileDeleted(state), WATCHER_CHECK_ERROR); + assert_int_equal(FileWatcherCheckFileDeleted(state), WATCHER_CHECK_ERROR); DestroyFileWatcherState(state); LogSetGlobalLevel(old_level); @@ -262,15 +440,15 @@ static void test_check_file_deleted_logs_error_once(void) /* The following failures are not logged again */ for (int i = 0; i < 3; i++) { - assert_int_equal(CheckFileDeleted(state), WATCHER_CHECK_ERROR); + assert_int_equal(FileWatcherCheckFileDeleted(state), WATCHER_CHECK_ERROR); } assert_int_equal(captured_err_count, 1); /* Once lstat() works again, the next failure is logged again */ assert_int_equal(unlink(loop), 0); - assert_int_equal(CheckFileDeleted(state), WATCHER_CHECK_NO_EVENT); + assert_int_equal(FileWatcherCheckFileDeleted(state), WATCHER_CHECK_NO_EVENT); assert_int_equal(symlink(loop, loop), 0); - assert_int_equal(CheckFileDeleted(state), WATCHER_CHECK_ERROR); + assert_int_equal(FileWatcherCheckFileDeleted(state), WATCHER_CHECK_ERROR); assert_int_equal(captured_err_count, 2); DestroyFileWatcherState(state); @@ -291,28 +469,28 @@ static void test_event_watcher_check_failure_reported_once(void) LogSetGlobalLevel(LOG_LEVEL_CRIT); WatcherRegistryInitialize(); - assert_true(WatcherRegister("failing-event", EVENT_FILE_DELETED, FileWatcherStateNew(path), &dummy_promise, 1)); + WatcherRegister(&expanded_promise, FileDeletedOptions(path), 1); int fd = -1; assert_true(EventWatcherInitialize(&fd)); /* The failing check is reported as a failure, not as an event */ - assert_true(WaitForQueuedEvent(fd, 5)); - handled_event_count = 0; - last_event_check = WATCHER_CHECK_NO_EVENT; - HandleEvents(fd, RecordEvent); + assert_true(WaitForQueuedEvent(fd, 50)); + ResetRecordedEvents(); + HandleEvents(fd); assert_int_equal(handled_event_count, 1); + AssertLastEventFor(&expanded_promise); assert_int_equal(last_event_check, WATCHER_CHECK_ERROR); /* The following failing checks (every second) are not reported again */ - assert_false(WaitForQueuedEvent(fd, 3)); + assert_false(WaitForQueuedEvent(fd, 30)); /* Once a check succeeds again, the next failure is reported again */ assert_int_equal(unlink(loop), 0); sleep(2); assert_int_equal(symlink(loop, loop), 0); - assert_true(WaitForQueuedEvent(fd, 5)); - HandleEvents(fd, RecordEvent); + assert_true(WaitForQueuedEvent(fd, 50)); + HandleEvents(fd); assert_int_equal(handled_event_count, 2); assert_int_equal(last_event_check, WATCHER_CHECK_ERROR); @@ -329,10 +507,14 @@ int main() { unit_test(test_registry_initialize_finalize), unit_test(test_watcher_register_single), - unit_test(test_watcher_register_duplicate_ignored), - unit_test(test_watcher_register_same_promiser_of_other_promise), + unit_test(test_watcher_register_iterations), + unit_test(test_watcher_register_identical_ignored), + unit_test(test_watcher_register_again_after_pause), unit_test(test_event_watcher_lifecycle), unit_test(test_event_watcher_pause_handles_queued_events), + unit_test(test_event_watcher_state_kept_across_pause), + unit_test(test_event_watcher_state_kept_identical_ignored), + unit_test(test_event_watcher_state_not_kept_for_other_promise), unit_test(test_check_file_deleted_error), unit_test(test_check_file_deleted_logs_error_once), unit_test(test_event_watcher_check_failure_reported_once),