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); 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),