[Fix] Adopt fuzzy storage for flexible backends

diff --git a/src/fuzzy_storage.c b/src/fuzzy_storage.c
index c49a9f277f871b1733771a34d95a1ace605e633a..bdb5f4aa4048e602cd5664553df51eabea3b9e0f 100644 (file)
--- a/src/fuzzy_storage.c
+++ b/src/fuzzy_storage.c
@@ -135,6 +135,7 @@ struct rspamd_fuzzy_storage_ctx {
        GQueue *updates_pending;
        struct rspamd_dns_resolver *resolver;
        struct rspamd_config *cfg;
+       struct rspamd_worker *worker;
 };
 
 enum fuzzy_cmd_type {
@@ -491,6 +492,16 @@ rspamd_fuzzy_updates_cb (gboolean success, void *ud)
                                g_queue_get_length (ctx->updates_pending));
        }
 
+       if (ctx->worker->wanna_die) {
+               /* Plan exit */
+               struct timeval tv;
+
+               tv.tv_sec = 0;
+               tv.tv_usec = 0;
+
+               event_base_loopexit (ctx->ev_base, &tv);
+       }
+
        g_slice_free1 (sizeof (*cbdata), cbdata);
 }
 
@@ -508,14 +519,6 @@ rspamd_fuzzy_process_updates_queue (struct rspamd_fuzzy_storage_ctx *ctx,
                cbdata->source = source;
                rspamd_fuzzy_backend_process_updates (ctx->backend, ctx->updates_pending,
                                source, rspamd_fuzzy_updates_cb, cbdata);
-
-
-       }
-       else if (ctx->updates_pending &&
-                       g_queue_get_length (ctx->updates_pending) > 0) {
-               msg_err ("cannot start transaction in fuzzy backend, "
-                               "%ud updates are still pending",
-                               g_queue_get_length (ctx->updates_pending));
        }
 }
 
@@ -2252,8 +2255,17 @@ start_fuzzy (struct rspamd_worker *worker)
                        "fuzzy",
                        NULL);
        ctx->peer_fd = -1;
+       ctx->worker = worker;
        double_to_tv (ctx->master_timeout, &ctx->master_io_tv);
 
+       ctx->resolver = dns_resolver_init (worker->srv->logger,
+                       ctx->ev_base,
+                       worker->srv->cfg);
+       rspamd_upstreams_library_config (worker->srv->cfg, ctx->cfg->ups_ctx,
+                       ctx->ev_base, ctx->resolver->r);
+       rspamd_redis_pool_config (worker->srv->cfg->redis_pool,
+                       worker->srv->cfg, ctx->ev_base);
+
        /*
         * Open DB and perform VACUUM
         */
@@ -2331,6 +2343,11 @@ start_fuzzy (struct rspamd_worker *worker)
        event_base_loop (ctx->ev_base, 0);
        rspamd_worker_block_signals ();
 
+       if (worker->index == 0 && g_queue_get_length (ctx->updates_pending) > 0) {
+               rspamd_fuzzy_process_updates_queue (ctx, local_db_name);
+               event_base_loop (ctx->ev_base, 0);
+       }
+
        rspamd_fuzzy_backend_close (ctx->backend);
        rspamd_log_close (worker->srv->logger);