From: Dmitriy Alekseev <1865999+dragoangel@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:00:52 +0000 (+0200) Subject: [Fix] Make the hyperscan cache usable with a Redis backend X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=e5d01170e2d67de6889bf3268f744983dbb7435b;p=thirdparty%2Frspamd.git [Fix] Make the hyperscan cache usable with a Redis backend Redis classifies GETEX as a write command, as it modifies the key expiry, but lua_hs_cache used it for the TTL refresh on read without setting is_write, so the request went to `read_servers`: with a split master/replicas setup every single cache load failed with `READONLY You can't write against a read only replica`, leaving all maps and multipatterns on the PCRE/ACISM fallback. The failure was easy to miss, as the check used to skip recompilation is EXISTS, a read command that replicas serve happily: startup stayed fast whilst no worker ever received a database. Use GETEX only when reads and writes are served by the same servers, otherwise read from the replicas and refresh the TTL by a separate EXPIRE against the master. Bail out early on store and delete when no write servers are configured, as lua_redis dereferences write_servers unconditionally, and sanitise the configured ttl, which reached Redis through tostring() and would mean `EXPIRE key 0` - dropping the very database just read - when set to anything non-positive. A nil reply is represented by the `redis{null}` sentinel: a truthy zero sized userdata whose metatable has a function as `__index`. It was passed to zstd_decompress, which failed the rspamd{text} check and then blew up inside the very code that formats the type mismatch, as it indexes `__index` expecting a table. Publish the sentinel as `rspamd_redis.null` with lua_redis.is_null() on top, so that a missing key is told from data by identity, and make that reporting robust: check the type of `__index`, resolve the class names before luaL_Buffer is initialised, and ask Lua for stack space beforehand, so the error being reported is not replaced by the one raised whilst describing it. Report why a hot-swap did not happen, as the reason used to be discarded and the only visible symptom was a `No such file or directory` warning from the local file fallback, which was attempted even for the backends that store nothing on disk. A plain cache miss is logged as info, whilst a real failure stays a warning. Load the databases asynchronously on worker startup as well, so a worker started after the hs_helper broadcast does not wait for the next one. Drop the trailing slash from the hs_cache_dir default: every consumer appends a separator itself, hence `dir//hash.hs` in paths and logs. That uncovers swapped arguments in rspamd_multipattern_try_save_hs(), which printed the separator before the directory and only produced a usable path because of that very slash. Finally, compile the regexp maps that were read after the fork. Their queue is process local and hs_helper never watches maps, so it only inherits what the main process has read before forking: a map read later on was queued in the process that read it and nobody ever compiled it, which is not an edge case, as rspamd_map_preload() gives up on a map unless every backend is a file or an HTTP one with a cached copy. A three thousand pattern map served over HTTP would therefore stay on PCRE for good. Let the process that read the map compile it: move the draining out of hs_helper into map_helpers, where the queue lives, and run it from the map periodic dtor in the primary controller, skipping the entries inherited from main, which hs_helper takes care of. The queue is walked by name, resolving the helper afresh at every step and dropping an entry only whilst its digest still matches, as the queue is an array that a map read can reallocate and a reload frees the helper behind it; a round is scheduled anew when an entry had to be left queued. Nothing is compiled twice, as the existing existence check is kept, and nothing is fetched twice either, as the database being loaded is remembered. Compiling only stores the database and a notification never comes back to its sender, so it is loaded locally as well, and a map is looked up in the cache the moment it is queued, which is how a worker picks up a database compiled by another process or instance. That notification now carries the digest of the compiled content instead of a name truncated to 64 bytes, which silently never matched for a map named after an URL. Signed-off-by: Dmitriy Alekseev <1865999+dragoangel@users.noreply.github.com> --- diff --git a/conf/options.inc b/conf/options.inc index 414fcb7d1c..56b34795e4 100644 --- a/conf/options.inc +++ b/conf/options.inc @@ -68,7 +68,7 @@ stats_file = "${DBDIR}/stats.ucl"; # Local networks local_addrs = [192.168.0.0/16, 10.0.0.0/8, 172.16.0.0/12, fd00::/8, 169.254.0.0/16, fe80::/10]; -hs_cache_dir = "${DBDIR}/"; +hs_cache_dir = "${DBDIR}"; # Timeout for messages processing (must be larger than any internal timeout used) task_timeout = 8s; diff --git a/lualib/lua_hs_cache.lua b/lualib/lua_hs_cache.lua index d60a1cf9cb..2fc8e9100c 100644 --- a/lualib/lua_hs_cache.lua +++ b/lualib/lua_hs_cache.lua @@ -425,6 +425,14 @@ end local redis_backend = {} redis_backend.__index = redis_backend +local redis_default_ttl = 86400 * 30 + +-- Redis expects an integer expire value, whilst ttl comes from the configuration +-- as a floating point number (e.g. `2592000.0`) +local function ttl_to_str(ttl) + return tostring(math.floor(ttl)) +end + function redis_backend.new(config) local self = setmetatable({}, redis_backend) @@ -455,7 +463,14 @@ function redis_backend.new(config) -- Config options can be in redis sub-section or at top level local opts = config.redis or config - self.default_ttl = opts.ttl or config.ttl or (86400 * 30) -- 30 days default + -- A non-positive ttl would mean `EXPIRE key 0`, i.e. dropping the cached + -- database, so refuse to use anything but a sane positive number here + self.default_ttl = tonumber(opts.ttl or config.ttl) or redis_default_ttl + if self.default_ttl < 1 then + logger.warnx(N, "invalid ttl %s configured for the redis hyperscan cache, use %s instead", + opts.ttl or config.ttl, redis_default_ttl) + self.default_ttl = redis_default_ttl + end self.refresh_ttl = (opts.refresh_ttl ~= false) and (config.refresh_ttl ~= false) self.use_compression = (opts.compression ~= false) and (config.compression ~= false) -- Use different default prefix for compressed (rspamd_zhs) vs uncompressed (rspamd_hs) @@ -472,6 +487,38 @@ function redis_backend:_get_key(cache_key, platform_id) return string.format("%s:%s:%s", self.prefix, platform_id, cache_key) end +-- Upstream lists can be updated in runtime (e.g. by sentinels), so writability +-- is checked for each request and not once on backend creation +function redis_backend:_can_write() + return self.redis_params ~= nil and self.redis_params.write_servers ~= nil +end + +-- True when reads are served by dedicated replicas: write commands must then be +-- issued as separate requests to the write servers +function redis_backend:_split_servers() + return self:_can_write() and + self.redis_params.read_servers_str ~= self.redis_params.write_servers_str +end + +-- Best effort TTL refresh performed on the write servers: EXPIRE is a write +-- command, so it cannot be issued as a part of a replica read +function redis_backend:_refresh_ttl(key) + local attrs = { + ev_base = self.redis_params.ev_base, + config = self.config, + is_write = true, + callback = function(err) + if err then + lua_util.debugm(N, self.config, "redis EXPIRE failed for key %s: %s", key, err) + else + lua_util.debugm(N, self.config, "redis refreshed TTL %s for key %s", self.default_ttl, key) + end + end + } + + lua_redis.request(self.redis_params, attrs, {'EXPIRE', key, ttl_to_str(self.default_ttl)}) +end + function redis_backend:exists(cache_key, platform_id, callback) local key = self:_get_key(cache_key, platform_id) @@ -508,20 +555,35 @@ function redis_backend:load(cache_key, platform_id, callback) return end - -- Use GETEX to refresh TTL on read if enabled - local req - if self.refresh_ttl then - lua_util.debugm(N, self.config, "redis GETEX (with TTL refresh %d) for key: %s", self.default_ttl, key) - req = {'GETEX', key, 'EX', tostring(self.default_ttl)} + -- GETEX mutates the key expiry, so it is a write command and must never be + -- sent to a read replica (it replies with `READONLY ...`). Use it only when + -- reads and writes go to the same servers, otherwise GET from the replicas + -- and refresh the TTL separately on the master. + local req, is_write + local refresh_separately = false + + if self.refresh_ttl and not self:_split_servers() then + lua_util.debugm(N, self.config, "redis GETEX (with TTL refresh %s) for key: %s", self.default_ttl, key) + req = {'GETEX', key, 'EX', ttl_to_str(self.default_ttl)} + is_write = true else lua_util.debugm(N, self.config, "redis GET for key: %s", key) req = {'GET', key} + refresh_separately = self.refresh_ttl and self:_can_write() end local attrs = { ev_base = self.redis_params.ev_base, config = self.config, + is_write = is_write, callback = function(err, data) + if lua_redis.is_null(data) then + data = nil + end + + if not err and data and refresh_separately then + self:_refresh_ttl(key) + end if err then lua_util.debugm(N, self.config, "redis GET failed for key %s: %s", key, err) callback(err, nil) @@ -575,6 +637,11 @@ function redis_backend:store(cache_key, platform_id, data, ttl, callback) return end + if not self:_can_write() then + callback("redis is configured for reading only") + return + end + lua_util.debugm(N, self.config, "redis SETEX for key: %s, original size: %d bytes, TTL: %d, compression: %s", key, #data, actual_ttl, self.use_compression and "enabled" or "disabled") @@ -607,7 +674,7 @@ function redis_backend:store(cache_key, platform_id, data, ttl, callback) end } - local req = {'SETEX', key, tostring(actual_ttl), store_data} + local req = {'SETEX', key, ttl_to_str(actual_ttl), store_data} lua_redis.request(self.redis_params, attrs, req) end @@ -619,6 +686,11 @@ function redis_backend:delete(cache_key, platform_id, callback) return end + if not self:_can_write() then + callback("redis is configured for reading only") + return + end + lua_util.debugm(N, self.config, "redis DEL for key: %s", key) local attrs = { diff --git a/lualib/lua_redis.lua b/lualib/lua_redis.lua index 750cdf144d..fccb47f328 100644 --- a/lualib/lua_redis.lua +++ b/lualib/lua_redis.lua @@ -2569,4 +2569,21 @@ exports.prepare_redis_setup = function(redis_params, opts, callback) return callback(nil) end +--[[[ +-- @function lua_redis.is_null(data) +-- Checks if a Redis reply carries no data: a nil reply is represented by a +-- sentinel userdata, which is truthy and cannot be compared to nil directly +-- @param {any} data data as passed to a Redis callback +-- @return {boolean} true if there is no data +--]] +exports.is_null = function(data) + if data == nil then + return true + end + + local rspamd_redis = require "rspamd_redis" + + return data == rspamd_redis.null +end + return exports diff --git a/src/hs_helper.c b/src/hs_helper.c index 2fa15949f5..6847633251 100644 --- a/src/hs_helper.c +++ b/src/hs_helper.c @@ -1079,205 +1079,6 @@ rspamd_hs_helper_compile_pending_multipatterns(struct hs_helper_ctx *ctx, rspamd_hs_helper_compile_pending_multipatterns_next(mpctx); } -/* - * Compile pending regexp maps that were queued during initialization - */ - -struct rspamd_hs_helper_remap_async_ctx { - struct hs_helper_ctx *ctx; - struct rspamd_worker *worker; - struct rspamd_regexp_map_pending *pending; - unsigned int count; - unsigned int idx; - gboolean compile_cb_called; - ref_entry_t ref; -}; - -static void rspamd_hs_helper_compile_pending_regexp_maps_next(struct rspamd_hs_helper_remap_async_ctx *rmctx); - -static void -rspamd_hs_helper_remap_async_ctx_dtor(void *p) -{ - struct rspamd_hs_helper_remap_async_ctx *rmctx = p; - rspamd_regexp_map_clear_pending(); - g_free(rmctx); -} - -static void -rspamd_hs_helper_remap_send_notification(struct hs_helper_ctx *ctx, - struct rspamd_worker *worker, - const char *name) -{ - struct rspamd_srv_command srv_cmd; - - memset(&srv_cmd, 0, sizeof(srv_cmd)); - srv_cmd.type = RSPAMD_SRV_REGEXP_MAP_LOADED; - rspamd_strlcpy(srv_cmd.cmd.re_map_loaded.name, name, - sizeof(srv_cmd.cmd.re_map_loaded.name)); - - rspamd_srv_send_command(worker, ctx->event_loop, &srv_cmd, -1, NULL, NULL); - msg_debug_hyperscan("sent regexp map loaded notification for '%s'", name); -} - -static void -rspamd_hs_helper_remap_compiled_cb(struct rspamd_regexp_map_helper *re_map, - gboolean success, - GError *err, - void *ud) -{ - struct rspamd_hs_helper_remap_async_ctx *rmctx = ud; - struct rspamd_regexp_map_pending *entry; - - (void) re_map; - - if (rmctx->compile_cb_called) { - REF_RELEASE(rmctx); - return; - } - rmctx->compile_cb_called = TRUE; - - entry = &rmctx->pending[rmctx->idx]; - rspamd_worker_set_busy(rmctx->worker, rmctx->ctx->event_loop, NULL); - - if (!success) { - msg_err("failed to compile regexp map '%s': %e", entry->name, err); - } - else { - rspamd_hs_helper_remap_send_notification(rmctx->ctx, rmctx->worker, entry->name); - } - - rmctx->idx++; - rspamd_hs_helper_compile_pending_regexp_maps_next(rmctx); - REF_RELEASE(rmctx); -} - -static void -rspamd_hs_helper_remap_exists_cb(gboolean success, - const unsigned char *data, - gsize len, - const char *error, - void *ud) -{ - struct rspamd_hs_helper_remap_async_ctx *rmctx = ud; - struct rspamd_regexp_map_pending *entry = &rmctx->pending[rmctx->idx]; - bool exists = (success && data == NULL && len == 1); - /* - * Save entry data before any operation that might trigger ev_run, - * as ev_run could process deferred timers that call - * rspamd_regexp_map_clear_pending() and free the pending array. - */ - struct rspamd_regexp_map_helper *re_map = entry->re_map; - const char *entry_name = entry->name; - - (void) error; - - if (exists) { - msg_debug_hyperscan("regexp map cache already exists for '%s', skipping compilation", entry_name); - rspamd_hs_helper_remap_send_notification(rmctx->ctx, rmctx->worker, entry_name); - rmctx->idx++; - rspamd_hs_helper_compile_pending_regexp_maps_next(rmctx); - REF_RELEASE(rmctx); - return; - } - - /* Need to compile+store */ - rspamd_worker_set_busy(rmctx->worker, rmctx->ctx->event_loop, "compile regexp map"); - /* - * DO NOT call ev_run() here - we're inside a Redis callback chain and - * ev_run can trigger Lua GC which may try to finalize lua_redis userdata - * while we're still processing. The busy notification will be sent on - * the next event loop iteration after this callback returns. - */ - rmctx->compile_cb_called = FALSE; - REF_RETAIN(rmctx); - rspamd_regexp_map_compile_hs_to_cache_async(re_map, rmctx->ctx->hs_dir, - rmctx->ctx->event_loop, - rspamd_hs_helper_remap_compiled_cb, rmctx); - /* Release the reference from exists_async callback */ - REF_RELEASE(rmctx); -} - -static void -rspamd_hs_helper_compile_pending_regexp_maps_next(struct rspamd_hs_helper_remap_async_ctx *rmctx) -{ - if (rmctx->worker->state != rspamd_worker_state_running) { - msg_debug_hyperscan("worker terminating, stopping regexp map compilation"); - goto done; - } - - if (rmctx->idx >= rmctx->count) { - goto done; - } - - struct rspamd_regexp_map_pending *entry = &rmctx->pending[rmctx->idx]; - msg_debug_hyperscan("processing regexp map '%s'", entry->name); - - if (rspamd_hs_cache_has_lua_backend()) { - char cache_key[rspamd_cryptobox_HASHBYTES * 2 + 1]; - rspamd_snprintf(cache_key, sizeof(cache_key), "%*xs", - (int) sizeof(entry->hash) / 2, entry->hash); - REF_RETAIN(rmctx); - rspamd_hs_cache_lua_exists_async(cache_key, entry->name, - rspamd_hs_helper_remap_exists_cb, rmctx); - return; - } - - /* File backend path: check if cache file exists */ - { - char fp[PATH_MAX]; - GError *err = NULL; - rspamd_snprintf(fp, sizeof(fp), "%s/%*xs.hsmc", rmctx->ctx->hs_dir, - (int) sizeof(entry->hash) / 2, entry->hash); - if (access(fp, R_OK) == 0) { - msg_debug_hyperscan("cache file %s already exists for regexp map '%s', skipping compilation", - fp, entry->name); - } - else { - rspamd_worker_set_busy(rmctx->worker, rmctx->ctx->event_loop, "compile regexp map"); - /* Flush the busy notification before blocking on compilation */ - ev_run(rmctx->ctx->event_loop, EVRUN_NOWAIT); - if (!rspamd_regexp_map_compile_hs_to_cache(entry->re_map, rmctx->ctx->hs_dir, &err)) { - msg_err("failed to compile regexp map '%s': %e", entry->name, err); - if (err) g_error_free(err); - } - rspamd_worker_set_busy(rmctx->worker, rmctx->ctx->event_loop, NULL); - } - - rspamd_hs_helper_remap_send_notification(rmctx->ctx, rmctx->worker, entry->name); - rmctx->idx++; - rspamd_hs_helper_compile_pending_regexp_maps_next(rmctx); - return; - } - -done: - REF_RELEASE(rmctx); -} - -static void -rspamd_hs_helper_compile_pending_regexp_maps(struct hs_helper_ctx *ctx, - struct rspamd_worker *worker) -{ - struct rspamd_regexp_map_pending *pending; - unsigned int count = 0; - - pending = rspamd_regexp_map_get_pending(&count); - if (pending == NULL || count == 0) { - msg_debug_hyperscan("no pending regexp map compilations"); - return; - } - - msg_debug_hyperscan("processing %ud pending regexp map compilations", count); - - struct rspamd_hs_helper_remap_async_ctx *rmctx = g_malloc0(sizeof(*rmctx)); - rmctx->ctx = ctx; - rmctx->worker = worker; - rmctx->pending = pending; - rmctx->count = count; - rmctx->idx = 0; - REF_INIT_RETAIN(rmctx, rspamd_hs_helper_remap_async_ctx_dtor); - - rspamd_hs_helper_compile_pending_regexp_maps_next(rmctx); -} #endif static gboolean @@ -1344,7 +1145,8 @@ rspamd_hs_helper_workers_spawned(struct rspamd_main *rspamd_main, rspamd_hs_helper_compile_pending_multipatterns(ctx, worker); /* Process pending regexp map compilations */ - rspamd_hs_helper_compile_pending_regexp_maps(ctx, worker); + rspamd_regexp_map_compile_pending_async(worker, ctx->event_loop, ctx->hs_dir, + RSPAMD_REGEXP_MAP_PENDING_DEFAULT); #endif if (attached_fd != -1) { @@ -1390,7 +1192,7 @@ start_hs_helper(struct rspamd_worker *worker) ctx->hs_dir = ctx->cfg->hs_cache_dir; } if (ctx->hs_dir == NULL) { - ctx->hs_dir = RSPAMD_DBDIR "/"; + ctx->hs_dir = RSPAMD_DBDIR; } /* Parse cache backend from config string */ diff --git a/src/libserver/cfg_utils.cxx b/src/libserver/cfg_utils.cxx index 4d2ff9c7fb..7467dd5847 100644 --- a/src/libserver/cfg_utils.cxx +++ b/src/libserver/cfg_utils.cxx @@ -334,7 +334,7 @@ rspamd_config_new(enum rspamd_config_init_flags flags) cfg->full_gc_iters = DEFAULT_GC_MAXITERS; /* Default hyperscan cache */ - cfg->hs_cache_dir = rspamd_mempool_strdup(cfg->cfg_pool, RSPAMD_DBDIR "/"); + cfg->hs_cache_dir = rspamd_mempool_strdup(cfg->cfg_pool, RSPAMD_DBDIR); if (!(flags & RSPAMD_CONFIG_INIT_SKIP_LUA)) { cfg->lua_state = (void *) rspamd_lua_init(flags & RSPAMD_CONFIG_INIT_WIPE_LUA_MEM); @@ -1020,7 +1020,7 @@ rspamd_config_post_load(struct rspamd_config *cfg, /* Try load Hypersan */ auto hs_ret = rspamd_re_cache_load_hyperscan(cfg->re_cache, - cfg->hs_cache_dir ? cfg->hs_cache_dir : RSPAMD_DBDIR "/", + cfg->hs_cache_dir ? cfg->hs_cache_dir : RSPAMD_DBDIR, true); if (hs_ret == RSPAMD_HYPERSCAN_LOAD_ERROR) { diff --git a/src/libserver/hs_cache_backend.c b/src/libserver/hs_cache_backend.c index 38ccfc2072..482a6af4e7 100644 --- a/src/libserver/hs_cache_backend.c +++ b/src/libserver/hs_cache_backend.c @@ -36,6 +36,8 @@ static struct rspamd_hs_cache_backend *global_hs_cache_backend = NULL; static lua_State *lua_backend_L = NULL; static int lua_backend_ref = LUA_NOREF; static const char *lua_backend_platform_id = NULL; +/* Name of the configured backend: file (default), redis or http */ +static char *lua_backend_name = NULL; static gboolean rspamd_hs_cache_try_init_lua_backend_with_opts(struct rspamd_config *cfg, @@ -134,6 +136,8 @@ rspamd_hs_cache_try_init_lua_backend_with_opts(struct rspamd_config *cfg, lua_pop(L, 1); rspamd_hs_cache_set_lua_backend(L, ref, platform_id); + g_free(lua_backend_name); + lua_backend_name = g_strdup(backend_name); lua_settop(L, err_idx - 1); msg_debug_hyperscan("initialized hyperscan cache backend: %s", backend_name); @@ -170,6 +174,8 @@ void rspamd_hs_cache_free_backend(void) lua_backend_L = NULL; lua_backend_ref = LUA_NOREF; lua_backend_platform_id = NULL; + g_free(lua_backend_name); + lua_backend_name = NULL; } void rspamd_hs_cache_set_lua_backend(lua_State *L, int ref, const char *platform_id) @@ -185,6 +191,30 @@ rspamd_hs_cache_has_lua_backend(void) return lua_backend_L != NULL && lua_backend_ref != LUA_NOREF; } +const char * +rspamd_hs_cache_backend_name(void) +{ + return lua_backend_name ? lua_backend_name : "file"; +} + +gboolean +rspamd_hs_cache_backend_is_file(void) +{ + return strcmp(rspamd_hs_cache_backend_name(), "file") == 0; +} + +gboolean +rspamd_hs_cache_error_is_miss(const char *error) +{ + if (error == NULL) { + /* Neither an error nor data: there is simply nothing cached */ + return TRUE; + } + + return strcmp(error, "not found") == 0 || + strcmp(error, "file not found") == 0; +} + gboolean rspamd_hs_cache_try_init_lua_backend(struct rspamd_config *cfg, struct ev_loop *ev_base) diff --git a/src/libserver/hs_cache_backend.h b/src/libserver/hs_cache_backend.h index ba3a7cd324..f59cdd5ebc 100644 --- a/src/libserver/hs_cache_backend.h +++ b/src/libserver/hs_cache_backend.h @@ -125,6 +125,29 @@ void rspamd_hs_cache_set_lua_backend(lua_State *L, int ref, const char *platform */ gboolean rspamd_hs_cache_has_lua_backend(void); +/** + * Get the name of the configured cache backend: `file`, `redis` or `http` + * @return backend name (never NULL, defaults to `file`) + */ +const char *rspamd_hs_cache_backend_name(void); + +/** + * Check if the configured backend stores hyperscan databases as local files, + * i.e. if reading them directly from `hs_cache_dir` makes any sense + * @return TRUE for the file backend + */ +gboolean rspamd_hs_cache_backend_is_file(void); + +/** + * Check if a load error merely means that nothing is cached yet, which is a + * normal condition: the database might not have been compiled so far, or a read + * replica might not have caught up with the master. Backends indicate this by + * returning `not found` (or `file not found` for the file one). + * @param error error message from a load callback (NULL means no data at all) + * @return TRUE if this is a cache miss rather than a backend failure + */ +gboolean rspamd_hs_cache_error_is_miss(const char *error); + /** * Initialize Lua HS cache backend in the current process using hs_helper worker * configuration (if configured and Lua is available). diff --git a/src/libserver/maps/map.c b/src/libserver/maps/map.c index 5541fe1177..98b59610ca 100644 --- a/src/libserver/maps/map.c +++ b/src/libserver/maps/map.c @@ -20,6 +20,7 @@ #include "config.h" #include "map.h" #include "map_private.h" +#include "map_helpers.h" #include "libserver/http/http_connection.h" #include "libserver/http/http_private.h" #include "rspamd.h" @@ -1397,6 +1398,21 @@ rspamd_map_periodic_dtor(struct map_periodic_cbdata *periodic) if (map->on_load_function) { map->on_load_function(map, map->on_load_ud); } + + /* + * A large regexp map is queued for hyperscan compilation by the read + * callback above. That queue is process local and hs_helper inherits + * merely the maps that have been read before the fork, so a map read + * later on would stay on the PCRE fallback forever unless we compile + * it here. Do it in the primary controller only, as every worker reads + * the same maps and would otherwise duplicate the work. + */ + if (rspamd_worker_is_primary_controller(map->wrk)) { + rspamd_regexp_map_compile_pending_async(map->wrk, map->event_loop, + map->cfg->hs_cache_dir ? map->cfg->hs_cache_dir : RSPAMD_DBDIR, + RSPAMD_REGEXP_MAP_PENDING_OWN_ONLY | + RSPAMD_REGEXP_MAP_PENDING_INSTALL); + } } else { /* Not modified */ @@ -3978,7 +3994,7 @@ void rspamd_map_trigger_hyperscan_compilation(struct rspamd_map *map) /* Use default settings for compilation */ rspamd_re_cache_compile_hyperscan_scoped_single(scope, scope_for_check, - map->cfg->hs_cache_dir ? map->cfg->hs_cache_dir : RSPAMD_DBDIR "/", + map->cfg->hs_cache_dir ? map->cfg->hs_cache_dir : RSPAMD_DBDIR, 1.0, /* max_time */ FALSE, /* silent */ worker->ctx ? ((struct rspamd_abstract_worker_ctx *) worker->ctx)->event_loop : NULL, diff --git a/src/libserver/maps/map_helpers.c b/src/libserver/maps/map_helpers.c index aafde1d7e1..32709c3782 100644 --- a/src/libserver/maps/map_helpers.c +++ b/src/libserver/maps/map_helpers.c @@ -16,6 +16,8 @@ #include "map_helpers.h" #include "map_private.h" +#include "libserver/rspamd_control.h" +#include "libserver/worker_util.h" #include "khash.h" #include "radix.h" #include "rspamd.h" @@ -23,12 +25,12 @@ #include "mempool_vars_internal.h" #include "rspamd_simdutf.h" #include "contrib/cdb/cdb.h" +#include "unix-std.h" #ifdef WITH_HYPERSCAN #include "hs.h" #include "hyperscan_tools.h" #include "hs_cache_backend.h" -#include "unix-std.h" #endif #ifndef WITH_PCRE2 #include @@ -1155,6 +1157,24 @@ rspamd_try_save_re_map_cache(struct rspamd_regexp_map_helper *re_map) return FALSE; } +static void +rspamd_regexp_map_cache_probe_cb(gboolean success, void *ud) +{ + char *name = ud; + + if (success) { + msg_info("hot-swapped queued regexp map '%s' to a hyperscan database " + "found in the cache backend", + name); + } + else { + msg_debug("no cached hyperscan database for the queued regexp map '%s'", + name); + } + + g_free(name); +} + #endif static void @@ -1296,6 +1316,22 @@ rspamd_re_map_finalize(struct rspamd_regexp_map_helper *re_map) map->name, re_map->regexps->len); rspamd_regexp_map_add_pending(re_map, map->name); + + /* + * Somebody might have compiled this very content already: another + * process of ours, or another instance sharing the cache backend. A + * notification is only broadcast when a database is compiled anew, so + * without looking into the cache here a map that has been read after + * that would keep using the fallback until the next compilation. + */ + if (rspamd_hs_cache_has_lua_backend() && map->event_loop != NULL && + !rspamd_worker_is_primary_controller(map->wrk)) { + rspamd_regexp_map_load_from_cache_async(re_map, + map->cfg->hs_cache_dir ? map->cfg->hs_cache_dir : RSPAMD_DBDIR, + map->event_loop, + rspamd_regexp_map_cache_probe_cb, + g_strdup(map->name)); + } } #endif } @@ -1910,6 +1946,7 @@ void rspamd_regexp_map_add_pending(struct rspamd_regexp_map_helper *re_map, struct rspamd_regexp_map_pending, i); if (strcmp(existing->name, name) == 0) { existing->re_map = re_map; + existing->queued_by = getpid(); rspamd_regexp_map_get_hash(re_map, existing->hash); msg_info_map("updated pending regexp map '%s' (%ud patterns) in compilation queue", @@ -1920,6 +1957,7 @@ void rspamd_regexp_map_add_pending(struct rspamd_regexp_map_helper *re_map, entry.re_map = re_map; entry.name = g_strdup(name); + entry.queued_by = getpid(); rspamd_regexp_map_get_hash(re_map, entry.hash); g_array_append_val(pending_regexp_maps, entry); @@ -1958,6 +1996,48 @@ void rspamd_regexp_map_clear_pending(void) pending_regexp_maps = NULL; } +void rspamd_regexp_map_remove_pending(const char *name) +{ + if (pending_regexp_maps == NULL || name == NULL) { + return; + } + + for (unsigned int i = 0; i < pending_regexp_maps->len; i++) { + struct rspamd_regexp_map_pending *entry; + + entry = &g_array_index(pending_regexp_maps, + struct rspamd_regexp_map_pending, i); + + if (strcmp(entry->name, name) == 0) { + g_free(entry->name); + g_array_remove_index(pending_regexp_maps, i); + + return; + } + } +} + +struct rspamd_regexp_map_helper * +rspamd_regexp_map_find_pending_by_hash(const unsigned char *hash) +{ + if (pending_regexp_maps == NULL || hash == NULL) { + return NULL; + } + + for (unsigned int i = 0; i < pending_regexp_maps->len; i++) { + struct rspamd_regexp_map_pending *entry; + + entry = &g_array_index(pending_regexp_maps, + struct rspamd_regexp_map_pending, i); + + if (memcmp(entry->hash, hash, rspamd_cryptobox_HASHBYTES) == 0) { + return entry->re_map; + } + } + + return NULL; +} + struct rspamd_regexp_map_helper * rspamd_regexp_map_find_pending(const char *name) { @@ -2302,13 +2382,55 @@ rspamd_regexp_map_load_from_cache(struct rspamd_regexp_map_helper *re_map, } struct rspamd_regexp_map_async_load_ctx { - struct rspamd_regexp_map_helper *re_map; + /* + * The helper is not kept here on purpose: a map can be re-read whilst the + * load is in flight and the old helper is then destroyed. The map itself + * lives as long as the configuration and knows the digest of its current + * content, which is enough to find the helper to install the database in + */ + struct rspamd_map *map; + uint64_t map_digest; void (*cb)(gboolean success, void *ud); void *ud; char *cache_dir; gboolean callback_processed; }; +/* + * Digests of the databases being loaded right now: a map is put in the queue by + * the read callback and announced by a notification, so without this the very + * same database would be fetched twice, and these are megabytes + */ +static GArray *inflight_map_loads = NULL; + +static gboolean +rspamd_regexp_map_load_inflight(uint64_t digest, gboolean add) +{ + if (inflight_map_loads == NULL) { + if (!add) { + return FALSE; + } + + inflight_map_loads = g_array_new(FALSE, FALSE, sizeof(uint64_t)); + } + + for (unsigned int i = 0; i < inflight_map_loads->len; i++) { + if (g_array_index(inflight_map_loads, uint64_t, i) == digest) { + if (!add) { + g_array_remove_index_fast(inflight_map_loads, i); + } + + return TRUE; + } + } + + if (add) { + g_array_append_val(inflight_map_loads, digest); + } + + return FALSE; +} + static void rspamd_regexp_map_async_load_cb(gboolean success, const unsigned char *data, @@ -2317,16 +2439,27 @@ rspamd_regexp_map_async_load_cb(gboolean success, void *ud) { struct rspamd_regexp_map_async_load_ctx *ctx = ud; - struct rspamd_map *map; + struct rspamd_map *map = ctx->map; + struct rspamd_regexp_map_helper *re_map = NULL; gboolean result = FALSE; if (ctx->callback_processed) { return; } ctx->callback_processed = TRUE; - map = ctx->re_map->map; + rspamd_regexp_map_load_inflight(ctx->map_digest, FALSE); - if (!success || data == NULL || len == 0) { + if (map->user_data != NULL && map->digest == ctx->map_digest) { + re_map = (struct rspamd_regexp_map_helper *) *map->user_data; + } + + if (re_map == NULL) { + /* The map has been re-read: the database we have asked for is stale */ + msg_info_map("skip the hyperscan database from the cache backend for %s, " + "the map has been re-read in the meantime", + map->name); + } + else if (!success || data == NULL || len == 0) { msg_warn_map("failed to load regexp map from cache backend: %s", error ? error : "no data"); } @@ -2338,23 +2471,23 @@ rspamd_regexp_map_async_load_cb(gboolean success, } else { /* Free old database if any */ - if (ctx->re_map->hs_db != NULL) { - rspamd_hyperscan_free(ctx->re_map->hs_db, true); - ctx->re_map->hs_db = NULL; + if (re_map->hs_db != NULL) { + rspamd_hyperscan_free(re_map->hs_db, true); + re_map->hs_db = NULL; } - if (ctx->re_map->hs_scratch != NULL) { - hs_free_scratch(ctx->re_map->hs_scratch); - ctx->re_map->hs_scratch = NULL; + if (re_map->hs_scratch != NULL) { + hs_free_scratch(re_map->hs_scratch); + re_map->hs_scratch = NULL; } - ctx->re_map->hs_db = rspamd_hyperscan_from_raw_db(db, NULL); + re_map->hs_db = rspamd_hyperscan_from_raw_db(db, NULL); - if (hs_alloc_scratch(rspamd_hyperscan_get_database(ctx->re_map->hs_db), - &ctx->re_map->hs_scratch) != HS_SUCCESS) { + if (hs_alloc_scratch(rspamd_hyperscan_get_database(re_map->hs_db), + &re_map->hs_scratch) != HS_SUCCESS) { msg_err_map("cannot allocate scratch space for hyperscan"); - rspamd_hyperscan_free(ctx->re_map->hs_db, true); - ctx->re_map->hs_db = NULL; + rspamd_hyperscan_free(re_map->hs_db, true); + re_map->hs_db = NULL; } else { msg_info_map("loaded hyperscan database from cache backend for %s", @@ -2388,17 +2521,433 @@ void rspamd_regexp_map_load_from_cache_async(struct rspamd_regexp_map_helper *re rspamd_snprintf(cache_key, sizeof(cache_key), "%*xs", (int) rspamd_cryptobox_HASHBYTES / 2, re_map->re_digest); + uint64_t map_digest; + + /* Same tag the map keeps for its current content, see rspamd_regexp_list_fin */ + memcpy(&map_digest, re_map->re_digest, sizeof(map_digest)); + + if (rspamd_regexp_map_load_inflight(map_digest, TRUE)) { + struct rspamd_map *map = re_map->map; + + msg_debug_map("the very database is already being loaded for %s", + map->name); + + if (cb) { + cb(FALSE, ud); + } + + return; + } + struct rspamd_regexp_map_async_load_ctx *ctx = g_malloc0(sizeof(*ctx)); - ctx->re_map = re_map; + ctx->map = re_map->map; ctx->cb = cb; ctx->ud = ud; ctx->cache_dir = g_strdup(cache_dir); + ctx->map_digest = map_digest; rspamd_hs_cache_lua_load_async(cache_key, re_map->map ? re_map->map->name : "regexp_map", rspamd_regexp_map_async_load_cb, ctx); } +/* + * Compile the queued regexp maps and notify the workers so that they hot-swap + * their databases. The queue is process local, so every process that has read + * a map on its own has to drive this: hs_helper only inherits what has been + * read before the fork. + */ +struct rspamd_regexp_map_pending_ctx { + struct rspamd_worker *worker; + struct ev_loop *event_loop; + char *cache_dir; + /* + * Own copy of the queued names: the queue itself is an array that a map + * read can reallocate whilst we are working, so pointers into it must not + * be kept, and the helper behind a name can be replaced by a map reload + */ + GPtrArray *names; + unsigned int idx; + unsigned int flags; + /* An entry has been left queued, so another round is due */ + gboolean requeue; + /* Digest of the version we are working on, to detect a map reload */ + unsigned char hash[rspamd_cryptobox_HASHBYTES]; + gboolean compile_cb_called; + ref_entry_t ref; +}; + +/* Only one drain at a time, as every map update triggers another attempt */ +static gboolean pending_regexp_maps_compiling = FALSE; + +static void rspamd_regexp_map_compile_pending_next(struct rspamd_regexp_map_pending_ctx *rmctx); + +/* Deferred restart of the draining, see the context destructor */ +struct rspamd_regexp_map_requeue { + struct rspamd_worker *worker; + struct ev_loop *event_loop; + char *cache_dir; + unsigned int flags; + ev_timer tm; +}; + +static void +rspamd_regexp_map_requeue_cb(EV_P_ ev_timer *w, int revents) +{ + struct rspamd_regexp_map_requeue *rq = (struct rspamd_regexp_map_requeue *) w->data; + + ev_timer_stop(EV_A_ w); + rspamd_regexp_map_compile_pending_async(rq->worker, rq->event_loop, + rq->cache_dir, rq->flags); + g_free(rq->cache_dir); + g_free(rq); +} + +static void +rspamd_regexp_map_pending_ctx_dtor(void *p) +{ + struct rspamd_regexp_map_pending_ctx *rmctx = p; + + pending_regexp_maps_compiling = FALSE; + + /* + * A map has been re-read whilst we were compiling it, so its new version is + * still queued: compile it at once instead of waiting for the next map + * update to trigger the draining again + */ + if (rmctx->requeue && rmctx->worker->state == rspamd_worker_state_running) { + struct rspamd_regexp_map_requeue *rq = g_malloc0(sizeof(*rq)); + + rq->worker = rmctx->worker; + rq->event_loop = rmctx->event_loop; + rq->cache_dir = g_strdup(rmctx->cache_dir); + rq->flags = rmctx->flags; + rq->tm.data = rq; + ev_timer_init(&rq->tm, rspamd_regexp_map_requeue_cb, 0.0, 0.0); + ev_timer_start(rq->event_loop, &rq->tm); + } + + g_ptr_array_free(rmctx->names, TRUE); + g_free(rmctx->cache_dir); + g_free(rmctx); +} + +/* + * Resolve the helper currently queued under this name and remember its digest. + * It has to be done afresh at every step, as a map reload replaces the helper + * and frees the old one whilst we are working + */ +static struct rspamd_regexp_map_helper * +rspamd_regexp_map_pending_resolve(struct rspamd_regexp_map_pending_ctx *rmctx, + const char *name) +{ + struct rspamd_regexp_map_helper *re_map; + + re_map = rspamd_regexp_map_find_pending(name); + + if (re_map != NULL) { + rspamd_regexp_map_get_hash(re_map, rmctx->hash); + } + + return re_map; +} + +/* + * Drop the entry unless the map has been re-read whilst we were compiling: in + * that case the queue holds a newer version that still has to be compiled + */ +static void +rspamd_regexp_map_pending_done_with(struct rspamd_regexp_map_pending_ctx *rmctx, + const char *name) +{ + struct rspamd_regexp_map_helper *re_map; + unsigned char hash[rspamd_cryptobox_HASHBYTES]; + + re_map = rspamd_regexp_map_find_pending(name); + + if (re_map == NULL) { + return; + } + + rspamd_regexp_map_get_hash(re_map, hash); + + if (memcmp(hash, rmctx->hash, sizeof(hash)) == 0) { + rspamd_regexp_map_remove_pending(name); + } + else { + msg_debug_hyperscan("regexp map '%s' has been re-read, leaving it queued", + name); + rmctx->requeue = TRUE; + } +} + +static void +rspamd_regexp_map_pending_notify(struct rspamd_regexp_map_pending_ctx *rmctx, + const char *name) +{ + struct rspamd_srv_command srv_cmd; + + memset(&srv_cmd, 0, sizeof(srv_cmd)); + srv_cmd.type = RSPAMD_SRV_REGEXP_MAP_LOADED; + memcpy(srv_cmd.cmd.re_map_loaded.digest, rmctx->hash, + sizeof(srv_cmd.cmd.re_map_loaded.digest)); + rspamd_strlcpy(srv_cmd.cmd.re_map_loaded.name, name, + sizeof(srv_cmd.cmd.re_map_loaded.name)); + + rspamd_srv_send_command(rmctx->worker, rmctx->event_loop, &srv_cmd, -1, + NULL, NULL); + msg_debug_hyperscan("sent regexp map loaded notification for '%s'", name); +} + +/* + * Load the database for our own copy of the map: compiling merely stores it, + * and the notification we have just sent never comes back to us, as the main + * process excludes the sender from the broadcast + */ +static void +rspamd_regexp_map_pending_install(struct rspamd_regexp_map_pending_ctx *rmctx, + const char *name) +{ + struct rspamd_regexp_map_helper *re_map; + + if (!(rmctx->flags & RSPAMD_REGEXP_MAP_PENDING_INSTALL)) { + return; + } + + re_map = rspamd_regexp_map_find_pending(name); + + if (re_map != NULL) { + rspamd_regexp_map_load_from_cache_async(re_map, rmctx->cache_dir, + rmctx->event_loop, + rspamd_regexp_map_cache_probe_cb, + g_strdup(name)); + } +} + +static void +rspamd_regexp_map_pending_compiled_cb(struct rspamd_regexp_map_helper *re_map, + gboolean success, + GError *err, + void *ud) +{ + struct rspamd_regexp_map_pending_ctx *rmctx = ud; + const char *name; + + (void) re_map; + + if (rmctx->compile_cb_called) { + REF_RELEASE(rmctx); + return; + } + rmctx->compile_cb_called = TRUE; + + name = g_ptr_array_index(rmctx->names, rmctx->idx); + rspamd_worker_set_busy(rmctx->worker, rmctx->event_loop, NULL); + + if (!success) { + msg_err("failed to compile regexp map '%s': %e", name, err); + } + else { + rspamd_regexp_map_pending_notify(rmctx, name); + rspamd_regexp_map_pending_install(rmctx, name); + } + + /* Done either way: a broken map would burn CPU on every map update */ + rspamd_regexp_map_pending_done_with(rmctx, name); + rmctx->idx++; + rspamd_regexp_map_compile_pending_next(rmctx); + REF_RELEASE(rmctx); +} + +static void +rspamd_regexp_map_pending_exists_cb(gboolean success, + const unsigned char *data, + gsize len, + const char *error, + void *ud) +{ + struct rspamd_regexp_map_pending_ctx *rmctx = ud; + const char *name = g_ptr_array_index(rmctx->names, rmctx->idx); + bool exists = (success && data == NULL && len == 1); + struct rspamd_regexp_map_helper *re_map; + + (void) error; + + if (exists) { + msg_debug_hyperscan("regexp map cache already exists for '%s', skipping compilation", + name); + rspamd_regexp_map_pending_notify(rmctx, name); + rspamd_regexp_map_pending_install(rmctx, name); + rspamd_regexp_map_pending_done_with(rmctx, name); + rmctx->idx++; + rspamd_regexp_map_compile_pending_next(rmctx); + REF_RELEASE(rmctx); + return; + } + + /* The event loop has been running whilst the check was in flight */ + re_map = rspamd_regexp_map_pending_resolve(rmctx, name); + + if (re_map == NULL) { + msg_debug_hyperscan("regexp map '%s' has left the queue", name); + rmctx->idx++; + rspamd_regexp_map_compile_pending_next(rmctx); + REF_RELEASE(rmctx); + return; + } + + /* Need to compile+store */ + rspamd_worker_set_busy(rmctx->worker, rmctx->event_loop, "compile regexp map"); + /* + * DO NOT call ev_run() here - we're inside a Redis callback chain and + * ev_run can trigger Lua GC which may try to finalize lua_redis userdata + * while we're still processing. The busy notification will be sent on + * the next event loop iteration after this callback returns. + */ + rmctx->compile_cb_called = FALSE; + REF_RETAIN(rmctx); + rspamd_regexp_map_compile_hs_to_cache_async(re_map, rmctx->cache_dir, + rmctx->event_loop, + rspamd_regexp_map_pending_compiled_cb, + rmctx); + /* Release the reference from exists_async callback */ + REF_RELEASE(rmctx); +} + +static void +rspamd_regexp_map_compile_pending_next(struct rspamd_regexp_map_pending_ctx *rmctx) +{ + struct rspamd_regexp_map_helper *re_map; + const char *name; + + while (rmctx->idx < rmctx->names->len) { + if (rmctx->worker->state != rspamd_worker_state_running) { + msg_debug_hyperscan("worker terminating, stopping regexp map compilation"); + goto done; + } + + name = g_ptr_array_index(rmctx->names, rmctx->idx); + re_map = rspamd_regexp_map_pending_resolve(rmctx, name); + + if (re_map == NULL) { + /* Compiled by an earlier drain or the map is gone */ + rmctx->idx++; + continue; + } + + msg_debug_hyperscan("processing regexp map '%s'", name); + + if (rspamd_hs_cache_has_lua_backend()) { + char cache_key[rspamd_cryptobox_HASHBYTES * 2 + 1]; + + rspamd_snprintf(cache_key, sizeof(cache_key), "%*xs", + (int) rspamd_cryptobox_HASHBYTES / 2, rmctx->hash); + REF_RETAIN(rmctx); + rspamd_hs_cache_lua_exists_async(cache_key, name, + rspamd_regexp_map_pending_exists_cb, rmctx); + return; + } + + /* File backend path: check if cache file exists */ + char fp[PATH_MAX]; + + rspamd_snprintf(fp, sizeof(fp), "%s/%*xs.hsmc", rmctx->cache_dir, + (int) rspamd_cryptobox_HASHBYTES / 2, rmctx->hash); + + if (access(fp, R_OK) == 0) { + msg_debug_hyperscan("cache file %s already exists for regexp map '%s', " + "skipping compilation", + fp, name); + } + else { + GError *err = NULL; + + rspamd_worker_set_busy(rmctx->worker, rmctx->event_loop, "compile regexp map"); + /* Flush the busy notification before blocking on compilation */ + ev_run(rmctx->event_loop, EVRUN_NOWAIT); + /* That could have re-read the map, so resolve the helper again */ + re_map = rspamd_regexp_map_pending_resolve(rmctx, name); + + if (re_map != NULL && + !rspamd_regexp_map_compile_hs_to_cache(re_map, rmctx->cache_dir, &err)) { + msg_err("failed to compile regexp map '%s': %e", name, err); + + if (err) { + g_error_free(err); + } + } + + rspamd_worker_set_busy(rmctx->worker, rmctx->event_loop, NULL); + } + + rspamd_regexp_map_pending_notify(rmctx, name); + rspamd_regexp_map_pending_done_with(rmctx, name); + rmctx->idx++; + } + +done: + REF_RELEASE(rmctx); +} + +void rspamd_regexp_map_compile_pending_async(struct rspamd_worker *worker, + struct ev_loop *event_loop, + const char *cache_dir, + unsigned int flags) +{ + struct rspamd_regexp_map_pending *pending; + unsigned int count = 0, i; + + if (worker == NULL || event_loop == NULL || cache_dir == NULL) { + return; + } + + if (pending_regexp_maps_compiling) { + msg_debug_hyperscan("regexp map compilation is already in progress"); + return; + } + + pending = rspamd_regexp_map_get_pending(&count); + + if (pending == NULL || count == 0) { + msg_debug_hyperscan("no pending regexp map compilations"); + return; + } + + msg_debug_hyperscan("processing %ud pending regexp map compilations", count); + + struct rspamd_regexp_map_pending_ctx *rmctx = g_malloc0(sizeof(*rmctx)); + rmctx->worker = worker; + rmctx->event_loop = event_loop; + rmctx->cache_dir = g_strdup(cache_dir); + rmctx->flags = flags; + rmctx->names = g_ptr_array_new_full(count, g_free); + + for (i = 0; i < count; i++) { + if ((flags & RSPAMD_REGEXP_MAP_PENDING_OWN_ONLY) && + pending[i].queued_by != getpid()) { + /* Inherited from the main process, hence hs_helper deals with it */ + msg_debug_hyperscan("skip regexp map '%s' queued before the fork", + pending[i].name); + continue; + } + + g_ptr_array_add(rmctx->names, g_strdup(pending[i].name)); + } + + if (rmctx->names->len == 0) { + g_ptr_array_free(rmctx->names, TRUE); + g_free(rmctx->cache_dir); + g_free(rmctx); + + return; + } + + pending_regexp_maps_compiling = TRUE; + REF_INIT_RETAIN(rmctx, rspamd_regexp_map_pending_ctx_dtor); + + rspamd_regexp_map_compile_pending_next(rmctx); +} + #else /* !WITH_HYPERSCAN */ gboolean @@ -2454,4 +3003,14 @@ void rspamd_regexp_map_load_from_cache_async(struct rspamd_regexp_map_helper *re } } +void rspamd_regexp_map_compile_pending_async(struct rspamd_worker *worker, + struct ev_loop *event_loop, + const char *cache_dir, + unsigned int flags) +{ + (void) worker; + (void) event_loop; + (void) cache_dir; +} + #endif /* WITH_HYPERSCAN */ diff --git a/src/libserver/maps/map_helpers.h b/src/libserver/maps/map_helpers.h index dea754e56b..08707e8c8e 100644 --- a/src/libserver/maps/map_helpers.h +++ b/src/libserver/maps/map_helpers.h @@ -39,6 +39,7 @@ struct rspamd_radix_map_helper; struct rspamd_hash_map_helper; struct rspamd_regexp_map_helper; struct ev_loop; +struct rspamd_worker; struct rspamd_cdb_map_helper; struct rspamd_map_helper_value; @@ -270,6 +271,18 @@ struct rspamd_regexp_map_pending { struct rspamd_regexp_map_helper *re_map; char *name; /* Map identifier for logging/IPC */ unsigned char hash[64]; /* Cache key hash (rspamd_cryptobox_HASHBYTES) */ + pid_t queued_by; /* Process that has read the map and queued it */ +}; + +/** + * Flags for rspamd_regexp_map_compile_pending_async() + */ +enum rspamd_regexp_map_pending_flags { + RSPAMD_REGEXP_MAP_PENDING_DEFAULT = 0, + /* Skip the entries inherited from the main process: hs_helper has them too */ + RSPAMD_REGEXP_MAP_PENDING_OWN_ONLY = (1u << 0), + /* Load the compiled database for own use, as hs_helper does not scan */ + RSPAMD_REGEXP_MAP_PENDING_INSTALL = (1u << 1), }; /** @@ -294,6 +307,40 @@ struct rspamd_regexp_map_pending *rspamd_regexp_map_get_pending(unsigned int *co */ void rspamd_regexp_map_clear_pending(void); +/** + * Remove a single processed entry from the pending queue. + * @param name identifier the map has been queued with + */ +void rspamd_regexp_map_remove_pending(const char *name); + +/** + * Compile all queued regexp maps to the hyperscan cache and notify the workers + * so that they hot-swap the databases, clearing the queue afterwards. Nothing + * happens if a compilation is already in progress or the queue is empty. + * + * The queue is process local, so it has to be driven by every process that + * reads maps on its own: hs_helper inherits only the maps that have been read + * before the fork, whilst maps read later are known just to the process that + * has read them. + * + * @param worker worker used to notify the main process + * @param event_loop event loop to run the asynchronous operations on + * @param cache_dir hyperscan cache directory (used by the file backend) + * @param flags see enum rspamd_regexp_map_pending_flags + */ +void rspamd_regexp_map_compile_pending_async(struct rspamd_worker *worker, + struct ev_loop *event_loop, + const char *cache_dir, + unsigned int flags); + +/** + * Find a pending regexp map by its digest, which identifies the very content + * that has been compiled, unlike a name. + * @param hash digest of rspamd_cryptobox_HASHBYTES bytes + * @return regexp map helper or NULL if not found + */ +struct rspamd_regexp_map_helper *rspamd_regexp_map_find_pending_by_hash(const unsigned char *hash); + /** * Find a pending regexp map by name. * @param name identifier diff --git a/src/libserver/rspamd_control.c b/src/libserver/rspamd_control.c index 55e4874c6e..68c4fbc9a2 100644 --- a/src/libserver/rspamd_control.c +++ b/src/libserver/rspamd_control.c @@ -1336,6 +1336,9 @@ rspamd_srv_handler(EV_P_ ev_io *w, int revents) /* Broadcast command to all workers */ memset(&wcmd, 0, sizeof(wcmd)); wcmd.type = RSPAMD_CONTROL_REGEXP_MAP_LOADED; + memcpy(wcmd.cmd.re_map_loaded.digest, + cmd.cmd.re_map_loaded.digest, + sizeof(wcmd.cmd.re_map_loaded.digest)); rspamd_strlcpy(wcmd.cmd.re_map_loaded.name, cmd.cmd.re_map_loaded.name, sizeof(wcmd.cmd.re_map_loaded.name)); diff --git a/src/libserver/rspamd_control.h b/src/libserver/rspamd_control.h index 32faa3cfb7..0f8e097b5d 100644 --- a/src/libserver/rspamd_control.h +++ b/src/libserver/rspamd_control.h @@ -91,7 +91,12 @@ struct rspamd_control_command { char name[64]; } mp_loaded; struct { - char name[64]; /* Map name */ + /* + * Digest identifies the database in the cache, whilst the name is + * merely for logging: a map name is an URL and has no sane bound + */ + unsigned char digest[64]; /* rspamd_cryptobox_HASHBYTES */ + char name[128]; } re_map_loaded; struct { char tag[32]; @@ -265,7 +270,12 @@ struct rspamd_srv_command { } mp_loaded; /* Sent when a regexp map hyperscan db is compiled */ struct { - char name[64]; /* Map name */ + /* + * Digest identifies the database in the cache, whilst the name is + * merely for logging: a map name is an URL and has no sane bound + */ + unsigned char digest[64]; /* rspamd_cryptobox_HASHBYTES */ + char name[128]; } re_map_loaded; struct { gboolean is_busy; diff --git a/src/libserver/worker_util.c b/src/libserver/worker_util.c index ed8fb67455..71b6b2f2fc 100644 --- a/src/libserver/worker_util.c +++ b/src/libserver/worker_util.c @@ -556,19 +556,32 @@ rspamd_prepare_worker(struct rspamd_worker *worker, const char *name, * PCRE for missing/stale patterns. Async notifications still handle updates. */ if (rspamd_hs_cache_has_lua_backend() && worker->srv->cfg->re_cache) { - const char *cache_dir = worker->srv->cfg->hs_cache_dir ? worker->srv->cfg->hs_cache_dir : RSPAMD_DBDIR "/"; - enum rspamd_hyperscan_status hs_status; + const char *cache_dir = worker->srv->cfg->hs_cache_dir ? worker->srv->cfg->hs_cache_dir : RSPAMD_DBDIR; - hs_status = rspamd_re_cache_load_hyperscan_scoped(worker->srv->cfg->re_cache, - cache_dir, true); - if (hs_status == RSPAMD_HYPERSCAN_LOADED_FULL) { - msg_info("worker startup: hyperscan fully loaded from cache"); - } - else if (hs_status == RSPAMD_HYPERSCAN_LOADED_PARTIAL) { - msg_info("worker startup: hyperscan partially loaded, waiting for hs_helper"); + if (!rspamd_hs_cache_backend_is_file()) { + /* + * Remote backends (redis, http) can only be queried asynchronously, + * there are no local files to read here + */ + msg_debug("worker startup: loading hyperscan from '%s' cache backend", + rspamd_hs_cache_backend_name()); + rspamd_re_cache_load_hyperscan_scoped_async(worker->srv->cfg->re_cache, + event_loop, cache_dir, true); } else { - msg_debug("worker startup: no hyperscan available yet, waiting for hs_helper"); + enum rspamd_hyperscan_status hs_status; + + hs_status = rspamd_re_cache_load_hyperscan_scoped(worker->srv->cfg->re_cache, + cache_dir, true); + if (hs_status == RSPAMD_HYPERSCAN_LOADED_FULL) { + msg_info("worker startup: hyperscan fully loaded from cache"); + } + else if (hs_status == RSPAMD_HYPERSCAN_LOADED_PARTIAL) { + msg_info("worker startup: hyperscan partially loaded, waiting for hs_helper"); + } + else { + msg_debug("worker startup: no hyperscan available yet, waiting for hs_helper"); + } } } #endif @@ -2057,13 +2070,19 @@ rspamd_worker_multipattern_async_loaded(gboolean success, void *ud) msg_debug_hyperscan("multipattern '%s' hot-swapped to hyperscan (backend)", cbd->name); } else { - /* Try file fallback if available */ - if (cbd->mp && cbd->cache_dir && rspamd_multipattern_load_from_cache(cbd->mp, cbd->cache_dir)) { + /* + * Try file fallback, but merely for the file backend: with redis or http + * backends databases are never stored in `hs_cache_dir`, so reading it + * would just fail with a misleading `No such file or directory` error + */ + if (rspamd_hs_cache_backend_is_file() && cbd->mp && cbd->cache_dir && + rspamd_multipattern_load_from_cache(cbd->mp, cbd->cache_dir)) { msg_debug_hyperscan("multipattern '%s' hot-swapped to hyperscan (file fallback)", cbd->name); } else { - msg_warn("failed to hot-swap multipattern '%s' to hyperscan, continuing with ACISM fallback", - cbd->name); + msg_warn("failed to hot-swap multipattern '%s' to hyperscan using '%s' cache backend, " + "continuing with ACISM fallback", + cbd->name, rspamd_hs_cache_backend_name()); } } @@ -2203,7 +2222,9 @@ rspamd_worker_regexp_map_ready(struct rspamd_main *rspamd_main, struct rspamd_control_reply rep; struct rspamd_regexp_map_helper *re_map; const char *name = cmd->cmd.re_map_loaded.name; - const char *cache_dir = worker->srv->cfg->hs_cache_dir; + const char *cache_dir = worker->srv->cfg->hs_cache_dir ? + worker->srv->cfg->hs_cache_dir : + RSPAMD_DBDIR; memset(&rep, 0, sizeof(rep)); rep.type = RSPAMD_CONTROL_REGEXP_MAP_LOADED; @@ -2211,7 +2232,12 @@ rspamd_worker_regexp_map_ready(struct rspamd_main *rspamd_main, msg_debug_hyperscan("received regexp map loaded notification for '%s'", name); - re_map = rspamd_regexp_map_find_pending(name); + /* + * Look the map up by the digest of the content that has been compiled: our + * own copy of that map might be of another version, and then there is + * nothing to install here + */ + re_map = rspamd_regexp_map_find_pending_by_hash(cmd->cmd.re_map_loaded.digest); if (re_map != NULL) { /* All file operations go through Lua backend */ @@ -2226,7 +2252,12 @@ rspamd_worker_regexp_map_ready(struct rspamd_main *rspamd_main, rep.reply.hs_loaded.status = 0; } else { - msg_warn("received regexp map notification for unknown '%s'", name); + /* + * Normal whenever our copy of that map is of another version: it gets + * its own database as soon as it is queued + */ + msg_info("no copy of regexp map '%s' with the compiled content here", + name); rep.reply.hs_loaded.status = ENOENT; } diff --git a/src/libutil/multipattern.c b/src/libutil/multipattern.c index b301753926..a3b9268719 100644 --- a/src/libutil/multipattern.c +++ b/src/libutil/multipattern.c @@ -536,8 +536,8 @@ rspamd_multipattern_try_save_hs(struct rspamd_multipattern *mp, return; } - rspamd_snprintf(fp, sizeof(fp), "%s%shs-XXXXXXXXXXXXX", G_DIR_SEPARATOR_S, - hs_cache_dir); + rspamd_snprintf(fp, sizeof(fp), "%s%shs-XXXXXXXXXXXXX", hs_cache_dir, + G_DIR_SEPARATOR_S); if ((fd = g_mkstemp_full(fp, O_CREAT | O_EXCL | O_WRONLY, 00644)) != -1) { /* Serialize with unified header format (magic, platform, CRC) */ @@ -1857,13 +1857,25 @@ rspamd_multipattern_load_from_cache_cb(gboolean success, } } } + else { + msg_debug("multipattern is not in the compiling state (%d), cannot hot-swap it", + (int) mp->state); + } + } + else if (rspamd_hs_cache_error_is_miss(err)) { + /* Not compiled yet or a read replica has not caught up with the master */ + msg_info("multipattern hyperscan database %s is not cached yet", + ctx->cache_key); } else { - (void) err; + /* Report the error, otherwise the failure is unexplainable */ + msg_warn("cannot load multipattern hyperscan database %s from cache backend: %s", + ctx->cache_key, err); } if (!ok && gerr) { - msg_debug("multipattern hs load failed: %s", gerr->message); + msg_warn("cannot deserialize multipattern hyperscan database %s: %s", + ctx->cache_key, gerr->message); } g_clear_error(&gerr); diff --git a/src/lua/lua_common.c b/src/lua/lua_common.c index dcc7cc70a5..812562177f 100644 --- a/src/lua/lua_common.c +++ b/src/lua/lua_common.c @@ -2007,6 +2007,32 @@ unsigned int rspamd_lua_table_size(lua_State *L, int tbl_pos) return tbl_size; } +/* + * Get a class name of an object for diagnostics purposes. `__index` is not + * necessarily a table: e.g. `redis{null}` has a function there, and indexing a + * function would raise an error instead of reporting the type mismatch we are + * about to complain about. Values are left on the stack to keep the returned + * string alive, a caller is expected to restore the stack afterwards. + */ +static const char * +rspamd_lua_class_name_diag(lua_State *L, int pos) +{ + const char *ret = NULL; + + if (lua_getmetatable(L, pos)) { + lua_pushstring(L, "__index"); + lua_rawget(L, -2); + + if (lua_istable(L, -1)) { + lua_pushstring(L, "class"); + lua_rawget(L, -2); + ret = lua_tostring(L, -1); + } + } + + return ret ? ret : "unknown"; +} + static void * rspamd_lua_check_udata_common(lua_State *L, int pos, const char *classname, gboolean fatal) @@ -2048,12 +2074,8 @@ err: if (fatal) { const char *actual_classname = NULL; - if (lua_type(L, pos) == LUA_TUSERDATA && lua_getmetatable(L, pos)) { - lua_pushstring(L, "__index"); - lua_gettable(L, -2); - lua_pushstring(L, "class"); - lua_gettable(L, -2); - actual_classname = lua_tostring(L, -1); + if (lua_type(L, pos) == LUA_TUSERDATA) { + actual_classname = rspamd_lua_class_name_diag(L, pos); } else { actual_classname = lua_typename(L, lua_type(L, pos)); @@ -2062,6 +2084,22 @@ err: luaL_Buffer buf; char tmp[512]; int r; + int nstack = MIN(top, 10); + const char *stack_classnames[10] = {NULL}; + + /* + * Resolve class names before the buffer is initialised: pushing anything + * on the stack whilst luaL_Buffer is in use would break it. Each lookup + * needs a few slots, and running out of them here would raise an error + * on top of the one we are reporting + */ + if (lua_checkstack(L, nstack * 3 + 8)) { + for (i = 1; i <= nstack; i++) { + if (lua_type(L, i) == LUA_TUSERDATA) { + stack_classnames[i - 1] = rspamd_lua_class_name_diag(L, i); + } + } + } luaL_buffinit(L, &buf); r = rspamd_snprintf(tmp, sizeof(tmp), @@ -2073,23 +2111,12 @@ err: r = rspamd_snprintf(tmp, sizeof(tmp), " stack(%d): ", top); luaL_addlstring(&buf, tmp, r); - for (i = 1; i <= MIN(top, 10); i++) { + for (i = 1; i <= nstack; i++) { if (lua_type(L, i) == LUA_TUSERDATA) { - const char *clsname; - - if (lua_getmetatable(L, i)) { - lua_pushstring(L, "__index"); - lua_gettable(L, -2); - lua_pushstring(L, "class"); - lua_gettable(L, -2); - clsname = lua_tostring(L, -1); - } - else { - clsname = lua_typename(L, lua_type(L, i)); - } - r = rspamd_snprintf(tmp, sizeof(tmp), "[%d: ud=%s] ", i, - clsname); + stack_classnames[i - 1] ? + stack_classnames[i - 1] : + "unknown"); luaL_addlstring(&buf, tmp, r); } else { diff --git a/src/lua/lua_redis.c b/src/lua/lua_redis.c index bd79a37019..34c6bdc6d4 100644 --- a/src/lua/lua_redis.c +++ b/src/lua/lua_redis.c @@ -1728,6 +1728,12 @@ lua_load_redis(lua_State *L) { lua_newtable(L); luaL_register(L, NULL, redislib_f); + /* + * Expose the sentinel used for a nil reply: it is a truthy userdata, so + * without it Lua has no way to tell a missing key from data + */ + lua_getfield(L, LUA_REGISTRYINDEX, "redis.null"); + lua_setfield(L, -2, "null"); return 1; }