[Fix] Propagate learn/stat errors more precisely

5 files changed, 92 insertions, 29 deletions

diff --git a/src/libstat/backends/backends.h b/src/libstat/backends/backends.h
index 32c925382..025e9bf34 100644
--- a/src/libstat/backends/backends.h
+++ b/src/libstat/backends/backends.h
@@ -39,15 +39,15 @@ struct rspamd_stat_backend {
 	gboolean (*process_tokens)(struct rspamd_task *task, GPtrArray *tokens,
 			gint id,
 			gpointer ctx);
-	void (*finalize_process)(struct rspamd_task *task,
+	gboolean (*finalize_process)(struct rspamd_task *task,
 			gpointer runtime, gpointer ctx);
 	gboolean (*learn_tokens)(struct rspamd_task *task, GPtrArray *tokens,
 			gint id,
 			gpointer ctx);
 	gulong (*total_learns)(struct rspamd_task *task,
 			gpointer runtime, gpointer ctx);
-	void (*finalize_learn)(struct rspamd_task *task,
-			gpointer runtime, gpointer ctx);
+	gboolean (*finalize_learn)(struct rspamd_task *task,
+			gpointer runtime, gpointer ctx, GError **err);
 	gulong (*inc_learns)(struct rspamd_task *task,
 			gpointer runtime, gpointer ctx);
 	gulong (*dec_learns)(struct rspamd_task *task,
@@ -68,15 +68,15 @@ struct rspamd_stat_backend {
 		gboolean rspamd_##name##_process_tokens (struct rspamd_task *task, \
                 GPtrArray *tokens, gint id, \
 				gpointer ctx); \
-		void rspamd_##name##_finalize_process (struct rspamd_task *task, \
+		gboolean rspamd_##name##_finalize_process (struct rspamd_task *task, \
 				gpointer runtime, \
 				gpointer ctx); \
 		gboolean rspamd_##name##_learn_tokens (struct rspamd_task *task, \
                 GPtrArray *tokens, gint id, \
 				gpointer ctx); \
-		void rspamd_##name##_finalize_learn (struct rspamd_task *task, \
+		gboolean rspamd_##name##_finalize_learn (struct rspamd_task *task, \
 				gpointer runtime, \
-				gpointer ctx); \
+				gpointer ctx, GError **err); \
 		gulong rspamd_##name##_total_learns (struct rspamd_task *task, \
 				gpointer runtime, \
 				gpointer ctx); \
diff --git a/src/libstat/backends/mmaped_file.c b/src/libstat/backends/mmaped_file.c
index b3861ca96..f87e9e6bd 100644
--- a/src/libstat/backends/mmaped_file.c
+++ b/src/libstat/backends/mmaped_file.c
@@ -1087,21 +1087,24 @@ rspamd_mmaped_file_get_stat (gpointer runtime,
 	return res;
 }
 
-void
+gboolean
 rspamd_mmaped_file_finalize_learn (struct rspamd_task *task, gpointer runtime,
-		gpointer ctx)
+		gpointer ctx, GError **err)
 {
 	rspamd_mmaped_file_t *mf = (rspamd_mmaped_file_t *)runtime;
 
 	if (mf != NULL) {
 		msync (mf->map, mf->len, MS_INVALIDATE | MS_ASYNC);
 	}
+
+	return TRUE;
 }
 
-void
+gboolean
 rspamd_mmaped_file_finalize_process (struct rspamd_task *task, gpointer runtime,
 		gpointer ctx)
 {
+	return TRUE;
 }
 
 gpointer
diff --git a/src/libstat/backends/redis_backend.c b/src/libstat/backends/redis_backend.c
index 3ae15bd4c..2f4baf3b8 100644
--- a/src/libstat/backends/redis_backend.c
+++ b/src/libstat/backends/redis_backend.c
@@ -72,6 +72,7 @@ struct redis_stat_runtime {
 	guint64 learned;
 	gint id;
 	gboolean has_event;
+	GError *err;
 };
 
 /* Used to get statistics from redis */
@@ -987,6 +988,11 @@ rspamd_redis_fin (gpointer data)
 		/* This calls for all callbacks pending */
 		redisAsyncFree (redis);
 	}
+
+	if (rt->err) {
+		g_error_free (rt->err);
+		rt->err = NULL;
+	}
 }
 
 static void
@@ -1007,6 +1013,11 @@ rspamd_redis_fin_learn (gpointer data)
 		/* This calls for all callbacks pending */
 		redisAsyncFree (redis);
 	}
+
+	if (rt->err) {
+		g_error_free (rt->err);
+		rt->err = NULL;
+	}
 }
 
 static void
@@ -1029,6 +1040,10 @@ rspamd_redis_timeout (gint fd, short what, gpointer d)
 		/* This calls for all callbacks pending */
 		redisAsyncFree (redis);
 	}
+
+	g_set_error (&rt->err, rspamd_redis_stat_quark (), ETIMEDOUT,
+			"error getting reply from redis server %s: timeout",
+			rspamd_upstream_name (rt->selected));
 }
 
 /* Called when we have connected to the redis server and got stats */
@@ -1076,6 +1091,10 @@ rspamd_redis_connected (redisAsyncContext *c, gpointer r, gpointer priv)
 		msg_err_task ("error getting reply from redis server %s: %s",
 				rspamd_upstream_name (rt->selected), c->errstr);
 		rspamd_upstream_fail (rt->selected);
+
+		g_set_error (&rt->err, rspamd_redis_stat_quark (), c->err,
+				"error getting reply from redis server %s: %s",
+				rspamd_upstream_name (rt->selected), c->errstr);
 	}
 
 }
@@ -1158,6 +1177,9 @@ rspamd_redis_processed (redisAsyncContext *c, gpointer r, gpointer priv)
 		if (rt->redis) {
 			rspamd_upstream_fail (rt->selected);
 		}
+		g_set_error (&rt->err, rspamd_redis_stat_quark (), c->err,
+				"cannot get values: error getting reply from redis server %s: %s",
+				rspamd_upstream_name (rt->selected), c->errstr);
 	}
 
 	if (rt->has_event) {
@@ -1184,6 +1206,10 @@ rspamd_redis_learned (redisAsyncContext *c, gpointer r, gpointer priv)
 		if (rt->redis) {
 			rspamd_upstream_fail (rt->selected);
 		}
+
+		g_set_error (&rt->err, rspamd_redis_stat_quark (), c->err,
+				"cannot get learned: error getting reply from redis server %s: %s",
+				rspamd_upstream_name (rt->selected), c->errstr);
 	}
 
 	if (rt->has_event) {
@@ -1573,7 +1599,7 @@ rspamd_redis_process_tokens (struct rspamd_task *task,
 	return FALSE;
 }
 
-void
+gboolean
 rspamd_redis_finalize_process (struct rspamd_task *task, gpointer runtime,
 		gpointer ctx)
 {
@@ -1589,6 +1615,15 @@ rspamd_redis_finalize_process (struct rspamd_task *task, gpointer runtime,
 		rt->redis = NULL;
 		redisAsyncFree (redis);
 	}
+
+	if (rt->err) {
+		g_error_free (rt->err);
+		rt->err = NULL;
+
+		return FALSE;
+	}
+
+	return TRUE;
 }
 
 gboolean
@@ -1750,9 +1785,9 @@ rspamd_redis_learn_tokens (struct rspamd_task *task, GPtrArray *tokens,
 }
 
 
-void
+gboolean
 rspamd_redis_finalize_learn (struct rspamd_task *task, gpointer runtime,
-		gpointer ctx)
+		gpointer ctx, GError **err)
 {
 	struct redis_stat_runtime *rt = REDIS_RUNTIME (runtime);
 	redisAsyncContext *redis;
@@ -1766,6 +1801,15 @@ rspamd_redis_finalize_learn (struct rspamd_task *task, gpointer runtime,
 		rt->redis = NULL;
 		redisAsyncFree (redis);
 	}
+
+	if (rt->err) {
+		g_propagate_error (err, rt->err);
+		rt->err = NULL;
+
+		return FALSE;
+	}
+
+	return TRUE;
 }
 
 gulong
diff --git a/src/libstat/backends/sqlite3_backend.c b/src/libstat/backends/sqlite3_backend.c
index 8682ca73f..78c22f7de 100644
--- a/src/libstat/backends/sqlite3_backend.c
+++ b/src/libstat/backends/sqlite3_backend.c
@@ -735,7 +735,7 @@ rspamd_sqlite3_process_tokens (struct rspamd_task *task,
 	return TRUE;
 }
 
-void
+gboolean
 rspamd_sqlite3_finalize_process (struct rspamd_task *task, gpointer runtime,
 		gpointer ctx)
 {
@@ -754,7 +754,7 @@ rspamd_sqlite3_finalize_process (struct rspamd_task *task, gpointer runtime,
 	rt->lang_id = -1;
 	rt->user_id = -1;
 
-	return;
+	return TRUE;
 }
 
 gboolean
@@ -819,9 +819,9 @@ rspamd_sqlite3_learn_tokens (struct rspamd_task *task, GPtrArray *tokens,
 	return TRUE;
 }
 
-void
+gboolean
 rspamd_sqlite3_finalize_learn (struct rspamd_task *task, gpointer runtime,
-		gpointer ctx)
+		gpointer ctx, GError **err)
 {
 	struct rspamd_stat_sqlite3_rt *rt = runtime;
 	struct rspamd_stat_sqlite3_db *bk;
@@ -852,8 +852,15 @@ rspamd_sqlite3_finalize_learn (struct rspamd_task *task, gpointer runtime,
 			&wal_checkpointed) != SQLITE_OK) {
 		msg_warn_task ("cannot commit checkpoint: %s",
 				sqlite3_errmsg (bk->sqlite));
+
+		g_set_error (err, rspamd_sqlite3_backend_quark (), 500,
+				"cannot commit checkpoint: %s",
+				sqlite3_errmsg (bk->sqlite));
+		return FALSE;
 	}
 #endif
+
+	return TRUE;
 }
 
 gulong
diff --git a/src/libstat/stat_process.c b/src/libstat/stat_process.c
index 886108d5d..e09673783 100644
--- a/src/libstat/stat_process.c
+++ b/src/libstat/stat_process.c
@@ -456,7 +456,7 @@ rspamd_stat_backends_process (struct rspamd_stat_ctx *st_ctx,
 	}
 }
 
-static void
+static gboolean
 rspamd_stat_backends_post_process (struct rspamd_stat_ctx *st_ctx,
 		struct rspamd_task *task)
 {
@@ -478,9 +478,13 @@ rspamd_stat_backends_post_process (struct rspamd_stat_ctx *st_ctx,
 		bk_run = g_ptr_array_index (task->stat_runtimes, i);
 
 		if (bk_run != NULL) {
-			st->backend->finalize_process (task, bk_run, st_ctx);
+			if (!st->backend->finalize_process (task, bk_run, st_ctx)) {
+				return FALSE;
+			}
 		}
 	}
+
+	return TRUE;
 }
 
 static void
@@ -611,8 +615,10 @@ rspamd_stat_classify (struct rspamd_task *task, lua_State *L, guint stage,
 	}
 	else if (stage == RSPAMD_TASK_STAGE_CLASSIFIERS_POST) {
 		/* Process classifiers */
-		rspamd_stat_backends_post_process (st_ctx, task);
-		rspamd_stat_classifiers_process (st_ctx, task);
+		if (rspamd_stat_backends_post_process (st_ctx, task)) {
+			rspamd_stat_classifiers_process (st_ctx, task);
+		}
+		/* Do not process classifiers on backend failures */
 	}
 
 	task->processed_stages |= stage;
@@ -938,8 +944,9 @@ end:
 static gboolean
 rspamd_stat_backends_post_learn (struct rspamd_stat_ctx *st_ctx,
 		struct rspamd_task *task,
-		 const gchar *classifier,
-		 gboolean spam)
+		const gchar *classifier,
+		gboolean spam,
+		GError **err)
 {
 	struct rspamd_classifier *cl;
 	struct rspamd_statfile *st;
@@ -957,11 +964,6 @@ rspamd_stat_backends_post_learn (struct rspamd_stat_ctx *st_ctx,
 			continue;
 		}
 
-		if (cl->cache) {
-			cache_run = cl->cache->runtime (task, cl->cachecf, TRUE);
-			cl->cache->learn (task, spam, cache_run);
-		}
-
 		if (cl->cfg->flags & RSPAMD_FLAG_CLASSIFIER_NO_BACKEND) {
 			res = TRUE;
 			continue;
@@ -979,7 +981,14 @@ rspamd_stat_backends_post_learn (struct rspamd_stat_ctx *st_ctx,
 				continue;
 			}
 
-			st->backend->finalize_learn (task, bk_run, st_ctx);
+			if (!st->backend->finalize_learn (task, bk_run, st_ctx, err)) {
+				return RSPAMD_STAT_PROCESS_ERROR;
+			}
+		}
+
+		if (cl->cache) {
+			cache_run = cl->cache->runtime (task, cl->cachecf, TRUE);
+			cl->cache->learn (task, spam, cache_run);
 		}
 	}
 
@@ -1029,7 +1038,7 @@ rspamd_stat_learn (struct rspamd_task *task,
 		}
 	}
 	else if (stage == RSPAMD_TASK_STAGE_LEARN_POST) {
-		if (!rspamd_stat_backends_post_learn (st_ctx, task, classifier, spam)) {
+		if (!rspamd_stat_backends_post_learn (st_ctx, task, classifier, spam, err)) {
 			return RSPAMD_STAT_PROCESS_ERROR;
 		}
 	}