# 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;
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)
-- 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)
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)
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)
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")
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
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 = {
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
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
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) {
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 */
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);
/* 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) {
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,
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);
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)
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)
*/
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).
#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"
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 */
/* 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,
#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"
#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 <pcre.h>
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
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
}
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",
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);
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)
{
}
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,
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");
}
}
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",
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
}
}
+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 */
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;
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),
};
/**
*/
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
/* 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));
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];
} 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;
* 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
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());
}
}
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;
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 */
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;
}
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) */
}
}
}
+ 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);
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)
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));
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),
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 {
{
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;
}
projects / thirdparty / rspamd.git / commitdiff
