* Add incr and decr commands to kvstorage.

* Add integers detection inside kvstorage.

9 files changed, 372 insertions, 53 deletions

diff --git a/src/kvstorage.c b/src/kvstorage.c
index 6f1f591ef..a2485375c 100644
--- a/src/kvstorage.c
+++ b/src/kvstorage.c
@@ -136,6 +136,7 @@ rspamd_kv_storage_insert (struct rspamd_kv_storage *storage, gpointer key,
 	gint 								steps = 0;
 	struct rspamd_kv_element           *elt;
 	gboolean							res = TRUE;
+	glong								longval;
 
 	/* Hard limit */
 	if (storage->max_memory > 0) {
@@ -177,12 +178,25 @@ rspamd_kv_storage_insert (struct rspamd_kv_storage *storage, gpointer key,
 	}
 
 	/* Insert elt to the cache */
-	elt = storage->cache->insert_func (storage->cache, key, data, len);
-	if (elt == NULL) {
-		return FALSE;
+
+	/* First of all check element for integer */
+	if (rspamd_strtol (data, len, &longval)) {
+		elt = storage->cache->insert_func (storage->cache, key, &longval, sizeof (glong));
+		if (elt == NULL) {
+			return FALSE;
+		}
+		else {
+			elt->flags |= KV_ELT_INTEGER;
+		}
 	}
-	elt->flags = flags;
-	elt->size = len;
+	else {
+		elt = storage->cache->insert_func (storage->cache, key, data, len);
+		if (elt == NULL) {
+			return FALSE;
+		}
+	}
+
+	elt->flags |= flags;
 	elt->expire = expire;
 	if (expire == 0) {
 		elt->flags |= KV_ELT_PERSISTENT;
@@ -245,6 +259,45 @@ rspamd_kv_storage_replace (struct rspamd_kv_storage *storage, gpointer key, stru
 	return res;
 }
 
+/** Increment value in kvstorage */
+gboolean
+rspamd_kv_storage_increment (struct rspamd_kv_storage *storage, gpointer key, glong *value)
+{
+	struct rspamd_kv_element			*elt = NULL, *belt;
+	glong								*lp;
+
+	/* First try to look at cache */
+	elt = storage->cache->lookup_func (storage->cache, key);
+
+	if (elt == NULL && storage->backend) {
+		belt = storage->backend->lookup_func (storage->backend, key);
+		if (belt) {
+			/* Put this element into cache */
+			if ((belt->flags & KV_ELT_INTEGER) != 0) {
+				rspamd_kv_storage_insert_internal (storage, ELT_KEY (belt), ELT_DATA (belt),
+					belt->size, belt->flags,
+					belt->expire, &elt);
+			}
+			if ((belt->flags & KV_ELT_DIRTY) == 0) {
+				g_free (belt);
+			}
+		}
+	}
+	if (elt && (elt->flags & KV_ELT_INTEGER) != 0) {
+		lp = &ELT_LONG (elt);
+		*lp += *value;
+		*value = *lp;
+		if (storage->backend) {
+			return storage->backend->replace_func (storage->backend, key, elt);
+		}
+		else {
+			return TRUE;
+		}
+	}
+
+	return FALSE;
+}
+
 /** Lookup an element inside kv storage */
 struct rspamd_kv_element*
 rspamd_kv_storage_lookup (struct rspamd_kv_storage *storage, gpointer key, time_t now)
@@ -578,6 +631,7 @@ rspamd_kv_hash_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->keylen = keylen;
 		elt->size = len;
 		elt->hash = rspamd_strcase_hash (key);
+		elt->flags = 0;
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
 		g_hash_table_insert (cache->hash, ELT_KEY (elt), elt);
@@ -596,6 +650,7 @@ rspamd_kv_hash_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->age = time (NULL);
 		elt->keylen = keylen;
 		elt->size = len;
+		elt->flags = 0;
 		elt->hash = rspamd_strcase_hash (key);
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
@@ -764,6 +819,7 @@ rspamd_kv_radix_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->keylen = keylen;
 		elt->size = len;
 		elt->hash = rkey;
+		elt->flags = 0;
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
 		radix32tree_insert (cache->tree, rkey, 0xffffffff, (uintptr_t)elt);
@@ -783,6 +839,7 @@ rspamd_kv_radix_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->keylen = keylen;
 		elt->size = len;
 		elt->hash = rkey;
+		elt->flags = 0;
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
 		radix32tree_insert (cache->tree, rkey, 0xffffffff, (uintptr_t)elt);
@@ -982,6 +1039,7 @@ rspamd_kv_judy_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->keylen = keylen;
 		elt->size = len;
 		elt->hash = rspamd_strcase_hash (key);
+		elt->flags = 0;
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
 		JHSI (pelt, cache->judy, ELT_KEY (elt), elt->keylen);
@@ -1002,6 +1060,7 @@ rspamd_kv_judy_insert (struct rspamd_kv_cache *c, gpointer key, gpointer value,
 		elt->keylen = keylen;
 		elt->size = len;
 		elt->hash = rspamd_strcase_hash (key);
+		elt->flags = 0;
 		memcpy (elt->data, key, keylen + 1);
 		memcpy (ELT_DATA (elt), value, len);
 		JHSI (pelt, cache->judy, ELT_KEY (elt), elt->keylen);
diff --git a/src/kvstorage.h b/src/kvstorage.h
index 5f48dbf39..586b93e6f 100644
--- a/src/kvstorage.h
+++ b/src/kvstorage.h
@@ -64,10 +64,12 @@ enum rspamd_kv_flags {
 	KV_ELT_PERSISTENT = 1 << 1,
 	KV_ELT_DIRTY = 1 << 2,
 	KV_ELT_OUSTED = 1 << 3,
-	KV_ELT_NEED_FREE = 1 << 4
+	KV_ELT_NEED_FREE = 1 << 4,
+	KV_ELT_INTEGER = 1 << 5
 };
 
 #define ELT_DATA(elt) (gchar *)(elt)->data + (elt)->keylen + 1
+#define ELT_LONG(elt) *((glong *)((elt)->data + (elt)->keylen + 1))
 #define ELT_KEY(elt) (gchar *)(elt)->data
 #define ELT_SIZE(elt) elt->size + sizeof(struct rspamd_kv_element) + elt->keylen + 1
 
@@ -141,6 +143,9 @@ gboolean rspamd_kv_storage_insert (struct rspamd_kv_storage *storage, gpointer k
 /** Replace an element in the kv storage */
 gboolean rspamd_kv_storage_replace (struct rspamd_kv_storage *storage, gpointer key, struct rspamd_kv_element *elt);
 
+/** Increment value in kvstorage */
+gboolean rspamd_kv_storage_increment (struct rspamd_kv_storage *storage, gpointer key, glong *value);
+
 /** Lookup an element inside kv storage */
 struct rspamd_kv_element* rspamd_kv_storage_lookup (struct rspamd_kv_storage *storage, gpointer key, time_t now);
 
diff --git a/src/kvstorage_server.c b/src/kvstorage_server.c
index 1b542458b..dd0c18b57 100644
--- a/src/kvstorage_server.c
+++ b/src/kvstorage_server.c
@@ -156,15 +156,52 @@ parse_kvstorage_command (struct kvstorage_session *session, gchar *c, guint len)
 		}
 	}
 	else if (len == 4) {
-		if (g_ascii_strncasecmp (c, "quit", 4) == 0) {
+		if ((c[0] == 'i' || c[0] == 'I')     &&
+				(c[1] == 'n' || c[1] == 'N') &&
+				(c[2] == 'c' || c[2] == 'C') &&
+				(c[3] == 'r' || c[3] == 'R')) {
+			session->command = KVSTORAGE_CMD_INCR;
+			session->arg_data.value = 1;
+		}
+		else if ((c[0] == 'd' || c[0] == 'D') &&
+				(c[1] == 'e' || c[1] == 'E')  &&
+				(c[2] == 'c' || c[2] == 'C')  &&
+				(c[3] == 'r' || c[3] == 'R')) {
+			session->command = KVSTORAGE_CMD_DECR;
+			session->arg_data.value = -1;
+		}
+		else if (g_ascii_strncasecmp (c, "quit", 4) == 0) {
 			session->command = KVSTORAGE_CMD_QUIT;
 		}
-		if (g_ascii_strncasecmp (c, "sync", 4) == 0 || g_ascii_strncasecmp (c, "save", 4) == 0) {
+		else if (g_ascii_strncasecmp (c, "sync", 4) == 0 || g_ascii_strncasecmp (c, "save", 4) == 0) {
 			session->command = KVSTORAGE_CMD_SYNC;
 		}
 	}
 	else if (len == 6) {
-		if (g_ascii_strncasecmp (c, "delete", 6) == 0) {
+		if ((c[0] == 'i' || c[0] == 'I')     &&
+				(c[1] == 'n' || c[1] == 'N') &&
+				(c[2] == 'c' || c[2] == 'C') &&
+				(c[3] == 'r' || c[3] == 'R') &&
+				(c[4] == 'b' || c[4] == 'B') &&
+				(c[5] == 'y' || c[5] == 'Y')) {
+			session->command = KVSTORAGE_CMD_INCR;
+			session->arg_data.value = 1;
+		}
+		else if ((c[0] == 'd' || c[0] == 'D') &&
+				(c[1] == 'e' || c[1] == 'E')  &&
+				(c[2] == 'c' || c[2] == 'C')  &&
+				(c[3] == 'r' || c[3] == 'R')  &&
+				(c[4] == 'b' || c[4] == 'B')  &&
+				(c[5] == 'y' || c[5] == 'Y')) {
+			session->command = KVSTORAGE_CMD_DECR;
+			session->arg_data.value = -1;
+		}
+		else if ((c[0] == 'd' || c[0] == 'D') &&
+				(c[1] == 'e' || c[1] == 'E')  &&
+				(c[2] == 'l' || c[2] == 'L')  &&
+				(c[3] == 'e' || c[3] == 'E')  &&
+				(c[4] == 't' || c[4] == 'T')  &&
+				(c[5] == 'e' || c[5] == 'E')) {
 			session->command = KVSTORAGE_CMD_DELETE;
 		}
 		else if (g_ascii_strncasecmp (c, "select", 6) == 0) {
@@ -271,7 +308,9 @@ parse_kvstorage_line (struct kvstorage_session *session, f_str_t *in)
 					session->key = memory_pool_alloc (session->pool, p - c + 1);
 					rspamd_strlcpy (session->key, c, p - c + 1);
 					/* Now we must select next state based on command */
-					if (session->command == KVSTORAGE_CMD_SET) {
+					if (session->command == KVSTORAGE_CMD_SET ||
+							session->command == KVSTORAGE_CMD_INCR ||
+							session->command == KVSTORAGE_CMD_DECR) {
 						/* Read flags */
 						state = 99;
 						if (is_redis) {
@@ -280,7 +319,12 @@ parse_kvstorage_line (struct kvstorage_session *session, f_str_t *in)
 							session->expire = 0;
 						}
 						else {
-							next_state = 3;
+							if (session->command == KVSTORAGE_CMD_SET) {
+								next_state = 3;
+							}
+							else {
+								next_state = 5;
+							}
 						}
 					}
 					else {
@@ -299,7 +343,13 @@ parse_kvstorage_line (struct kvstorage_session *session, f_str_t *in)
 				if (g_ascii_isspace (*p)) {
 					session->flags = strtoul (c, NULL, 10);
 					state = 99;
-					next_state = 4;
+					if (session->command == KVSTORAGE_CMD_SET) {
+						next_state = 4;
+					}
+					else {
+						/* INCR and DECR */
+						next_state = 5;
+					}
 				}
 				else {
 					return FALSE;
@@ -323,13 +373,29 @@ parse_kvstorage_line (struct kvstorage_session *session, f_str_t *in)
 			}
 			break;
 		case 5:
-			/* Read size */
+			/* Read size or incr/decr values */
 			if (g_ascii_isdigit (*p)) {
 				p ++;
 			}
 			else {
-				if (g_ascii_isspace (*p) || end == p) {
-					session->length = strtoul (c, NULL, 10);
+				if (g_ascii_isspace (*p) || p >= end - 1) {
+					if (session->command == KVSTORAGE_CMD_SET) {
+						session->arg_data.length = strtoul (c, NULL, 10);
+					}
+					else {
+						if (p != c) {
+							session->arg_data.value = strtoul (c, NULL, 10);
+							if (session->command == KVSTORAGE_CMD_DECR) {
+								session->arg_data.value = -session->arg_data.value;
+							}
+						}
+						else if (session->command == KVSTORAGE_CMD_INCR) {
+							session->arg_data.value = 1;
+						}
+						else {
+							session->arg_data.value = -1;
+						}
+					}
 					state = 100;
 				}
 				else {
@@ -395,13 +461,15 @@ static gboolean
 kvstorage_process_command (struct kvstorage_session *session, gboolean is_redis)
 {
 	gint								 r;
-	gchar								 outbuf[BUFSIZ];
+	gchar								 outbuf[BUFSIZ], intbuf[sizeof ("9223372036854775807")];
 	gboolean							 res;
 	struct rspamd_kv_element			*elt;
+	guint								 eltlen;
+	glong								 longval;
 
 	if (session->command == KVSTORAGE_CMD_SET) {
 		session->state = KVSTORAGE_STATE_READ_DATA;
-		rspamd_set_dispatcher_policy (session->dispather, BUFFER_CHARACTER, session->length);
+		rspamd_set_dispatcher_policy (session->dispather, BUFFER_CHARACTER, session->arg_data.length);
 	}
 	else if (session->command == KVSTORAGE_CMD_GET) {
 		g_static_rw_lock_reader_lock (&session->cf->storage->rwlock);
@@ -419,23 +487,38 @@ kvstorage_process_command (struct kvstorage_session *session, gboolean is_redis)
 		}
 		else {
 			session->elt = elt;
+			if (elt->flags & KV_ELT_INTEGER) {
+				eltlen = rspamd_snprintf (intbuf, sizeof (intbuf), "%l", ELT_LONG (elt));
+
+			}
+			else {
+				eltlen = elt->size;
+			}
 
 			if (!is_redis) {
 				r = rspamd_snprintf (outbuf, sizeof (outbuf), "VALUE %s %ud %ud" CRLF,
-						ELT_KEY (elt), elt->flags, elt->size);
+						ELT_KEY (elt), elt->flags, eltlen);
 			}
 			else {
 				r = rspamd_snprintf (outbuf, sizeof (outbuf), "$%ud" CRLF,
-						elt->size);
+						eltlen);
 			}
 			if (!rspamd_dispatcher_write (session->dispather, outbuf,
 					r, TRUE, FALSE)) {
 				g_static_rw_lock_reader_unlock (&session->cf->storage->rwlock);
 				return FALSE;
 			}
-			if (!rspamd_dispatcher_write (session->dispather, ELT_DATA(elt), elt->size, TRUE, TRUE)) {
-				g_static_rw_lock_reader_unlock (&session->cf->storage->rwlock);
-				return FALSE;
+			if (elt->flags & KV_ELT_INTEGER) {
+				if (!rspamd_dispatcher_write (session->dispather, intbuf, eltlen, TRUE, TRUE)) {
+					g_static_rw_lock_reader_unlock (&session->cf->storage->rwlock);
+					return FALSE;
+				}
+			}
+			else {
+				if (!rspamd_dispatcher_write (session->dispather, ELT_DATA(elt), eltlen, TRUE, TRUE)) {
+					g_static_rw_lock_reader_unlock (&session->cf->storage->rwlock);
+					return FALSE;
+				}
 			}
 			if (!is_redis) {
 				res = rspamd_dispatcher_write (session->dispather, CRLF "END" CRLF,
@@ -482,6 +565,36 @@ kvstorage_process_command (struct kvstorage_session *session, gboolean is_redis)
 			}
 		}
 	}
+	else if (session->command == KVSTORAGE_CMD_INCR || session->command == KVSTORAGE_CMD_DECR) {
+		g_static_rw_lock_writer_lock (&session->cf->storage->rwlock);
+		longval = session->arg_data.value;
+		if (!rspamd_kv_storage_increment (session->cf->storage, session->key, &longval)) {
+			g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
+			if (!is_redis) {
+				return rspamd_dispatcher_write (session->dispather, ERROR_NOT_FOUND,
+						sizeof (ERROR_NOT_FOUND) - 1, FALSE, TRUE);
+			}
+			else {
+				return rspamd_dispatcher_write (session->dispather, "-ERR not found" CRLF,
+						sizeof ("-ERR not found" CRLF) - 1, FALSE, TRUE);
+			}
+		}
+		else {
+			g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
+			if (!is_redis) {
+				r = rspamd_snprintf (outbuf, sizeof (outbuf), "%l" CRLF,
+						longval);
+			}
+			else {
+				r = rspamd_snprintf (outbuf, sizeof (outbuf), "$%l" CRLF,
+						longval);
+			}
+			if (!rspamd_dispatcher_write (session->dispather, outbuf,
+					r, FALSE, FALSE)) {
+				return FALSE;
+			}
+		}
+	}
 	else if (session->command == KVSTORAGE_CMD_SYNC) {
 		if (session->cf->storage->backend == NULL || session->cf->storage->backend->sync_func == NULL) {
 			if (!is_redis) {
@@ -586,6 +699,9 @@ kvstorage_check_argnum (struct kvstorage_session *session)
 		return session->argc == 1;
 	case KVSTORAGE_CMD_SET:
 		return session->argc == 3;
+	case KVSTORAGE_CMD_INCR:
+	case KVSTORAGE_CMD_DECR:
+		return session->argc == 1 || session->argc == 2;
 	default:
 		return session->argc == 2;
 	}
@@ -682,9 +798,16 @@ kvstorage_read_socket (f_str_t * in, void *arg)
 							r, FALSE, TRUE);
 				}
 				else {
-					session->argnum ++;
-					session->state = KVSTORAGE_STATE_READ_ARGLEN;
-					rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
+					if (session->argnum == session->argc - 1) {
+						session->state = KVSTORAGE_STATE_READ_CMD;
+						rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
+						return kvstorage_process_command (session, TRUE);
+					}
+					else {
+						session->argnum ++;
+						session->state = KVSTORAGE_STATE_READ_ARGLEN;
+						rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
+					}
 				}
 			}
 		}
@@ -693,6 +816,7 @@ kvstorage_read_socket (f_str_t * in, void *arg)
 				/* This argument is a key for normal command */
 				session->key = memory_pool_fstrdup (session->pool, in);
 				if (session->argnum == session->argc - 1) {
+					session->state = KVSTORAGE_STATE_READ_CMD;
 					rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
 					return kvstorage_process_command (session, TRUE);
 				}
@@ -704,6 +828,7 @@ kvstorage_read_socket (f_str_t * in, void *arg)
 			}
 			else {
 				/* Special case for select command */
+				session->state = KVSTORAGE_STATE_READ_CMD;
 				rspamd_strlcpy (outbuf, in->begin, MIN (sizeof (outbuf), in->len));
 				session->id = strtoul (outbuf, NULL, 10);
 				rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
@@ -712,19 +837,30 @@ kvstorage_read_socket (f_str_t * in, void *arg)
 		}
 		else if (session->argnum == 2) {
 			/* We get datablock for set command */
-			session->state = KVSTORAGE_STATE_READ_CMD;
-			rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
-			g_static_rw_lock_writer_lock (&session->cf->storage->rwlock);
-			if (rspamd_kv_storage_insert (session->cf->storage, session->key, in->begin, in->len,
-					session->flags, session->expire)) {
-				g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
-				return rspamd_dispatcher_write (session->dispather, "+OK" CRLF,
-						sizeof ("+OK" CRLF) - 1, FALSE, TRUE);
+			if (session->command == KVSTORAGE_CMD_SET) {
+				session->state = KVSTORAGE_STATE_READ_CMD;
+				rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
+				g_static_rw_lock_writer_lock (&session->cf->storage->rwlock);
+				if (rspamd_kv_storage_insert (session->cf->storage, session->key, in->begin, in->len,
+						session->flags, session->expire)) {
+					g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
+					return rspamd_dispatcher_write (session->dispather, "+OK" CRLF,
+							sizeof ("+OK" CRLF) - 1, FALSE, TRUE);
+				}
+				else {
+					g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
+					return rspamd_dispatcher_write (session->dispather, "-ERR not stored" CRLF,
+							sizeof ("-ERR not stored" CRLF) - 1, FALSE, TRUE);
+				}
 			}
 			else {
-				g_static_rw_lock_writer_unlock (&session->cf->storage->rwlock);
-				return rspamd_dispatcher_write (session->dispather, "-ERR not stored" CRLF,
-						sizeof ("-ERR not stored" CRLF) - 1, FALSE, TRUE);
+				session->state = KVSTORAGE_STATE_READ_CMD;
+				rspamd_strtol (in->begin, in->len, &session->arg_data.value);
+				if (session->command == KVSTORAGE_CMD_DECR) {
+					session->arg_data.value = -session->arg_data.value;
+				}
+				rspamd_set_dispatcher_policy (session->dispather, BUFFER_LINE, -1);
+				return kvstorage_process_command (session, TRUE);
 			}
 		}
 		break;
diff --git a/src/kvstorage_server.h b/src/kvstorage_server.h
index a7c8bd8a6..ed4ce6eb0 100644
--- a/src/kvstorage_server.h
+++ b/src/kvstorage_server.h
@@ -69,6 +69,8 @@ struct kvstorage_session {
 		KVSTORAGE_CMD_DELETE,
 		KVSTORAGE_CMD_SYNC,
 		KVSTORAGE_CMD_SELECT,
+		KVSTORAGE_CMD_INCR,
+		KVSTORAGE_CMD_DECR,
 		KVSTORAGE_CMD_QUIT
 	} command;
 	guint id;
@@ -83,7 +85,10 @@ struct kvstorage_session {
 	gint sock;
 	guint flags;
 	guint expire;
-	guint length;
+	union {
+		glong value;
+		guint length;
+	} arg_data;
 	time_t now;
 };
 
diff --git a/src/kvstorage_sqlite.c b/src/kvstorage_sqlite.c
index 3f4fe6979..52181e633 100644
--- a/src/kvstorage_sqlite.c
+++ b/src/kvstorage_sqlite.c
@@ -254,13 +254,24 @@ rspamd_sqlite_insert (struct rspamd_kv_backend *backend, gpointer key, struct rs
 		return FALSE;
 	}
 
-	op = g_slice_alloc (sizeof (struct sqlite_op));
-	op->op = SQLITE_OP_INSERT;
-	op->elt = elt;
-	elt->flags |= KV_ELT_DIRTY;
+	if ((op = g_hash_table_lookup (db->ops_hash, key)) != NULL) {
+		/* We found another op with such key in this queue */
+		if (op->op == SQLITE_OP_DELETE || (op->elt->flags & KV_ELT_NEED_FREE) != 0) {
+			/* Also clean memory */
+			g_slice_free1 (ELT_SIZE (op->elt), op->elt);
+		}
+		op->op = SQLITE_OP_INSERT;
+		op->elt = elt;
+	}
+	else {
+		op = g_slice_alloc (sizeof (struct sqlite_op));
+		op->op = SQLITE_OP_INSERT;
+		op->elt = elt;
+		elt->flags |= KV_ELT_DIRTY;
 
-	g_queue_push_head (db->ops_queue, op);
-	g_hash_table_insert (db->ops_hash, ELT_KEY (elt), op);
+		g_queue_push_head (db->ops_queue, op);
+		g_hash_table_insert (db->ops_hash, ELT_KEY (elt), op);
+	}
 
 	if (db->sync_ops > 0 && g_queue_get_length (db->ops_queue) >= db->sync_ops) {
 		return sqlite_process_queue (backend);
@@ -278,14 +289,24 @@ rspamd_sqlite_replace (struct rspamd_kv_backend *backend, gpointer key, struct r
 	if (!db->initialized) {
 		return FALSE;
 	}
+	if ((op = g_hash_table_lookup (db->ops_hash, key)) != NULL) {
+		/* We found another op with such key in this queue */
+		if (op->op == SQLITE_OP_DELETE || (op->elt->flags & KV_ELT_NEED_FREE) != 0) {
+			/* Also clean memory */
+			g_slice_free1 (ELT_SIZE (op->elt), op->elt);
+		}
+		op->op = SQLITE_OP_REPLACE;
+		op->elt = elt;
+	}
+	else {
+		op = g_slice_alloc (sizeof (struct sqlite_op));
+		op->op = SQLITE_OP_REPLACE;
+		op->elt = elt;
+		elt->flags |= KV_ELT_DIRTY;
 
-	op = g_slice_alloc (sizeof (struct sqlite_op));
-	op->op = SQLITE_OP_REPLACE;
-	op->elt = elt;
-	elt->flags |= KV_ELT_DIRTY;
-
-	g_queue_push_head (db->ops_queue, op);
-	g_hash_table_insert (db->ops_hash, ELT_KEY (elt), op);
+		g_queue_push_head (db->ops_queue, op);
+		g_hash_table_insert (db->ops_hash, ELT_KEY (elt), op);
+	}
 
 	if (db->sync_ops > 0 && g_queue_get_length (db->ops_queue) >= db->sync_ops) {
 		return sqlite_process_queue (backend);
diff --git a/src/main.c b/src/main.c
index 51f0a881f..da6d76e48 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1050,7 +1050,8 @@ main (gint argc, gchar **argv, gchar **env)
 		print_signals_info ();
 #endif
 		if (do_terminate) {
-			msg_debug ("catch termination signal, waiting for childs");
+			do_terminate = 0;
+			msg_info ("catch termination signal, waiting for childs");
 			pass_signal_worker (rspamd_main->workers, SIGTERM);
 			break;
 		}
diff --git a/src/printf.c b/src/printf.c
index 54919a5c1..3ab5be3df 100644
--- a/src/printf.c
+++ b/src/printf.c
@@ -468,9 +468,9 @@ rspamd_vsnprintf (gchar *buf, glong max, const gchar *fmt, va_list args)
 
 			case 'l':
 				if (sign) {
-					i64 = (gint64) va_arg(args, long);
+					i64 = (gint64) va_arg(args, glong);
 				} else {
-					ui64 = (guint64) va_arg(args, guint32);
+					ui64 = (guint64) va_arg(args, gulong);
 				}
 				break;
 
diff --git a/src/util.c b/src/util.c
index 570427ef7..b39adfec1 100644
--- a/src/util.c
+++ b/src/util.c
@@ -1504,5 +1504,87 @@ rspamd_strncasestr (const gchar *s, const gchar *find, gint len)
 }
 
 /*
+ * Try to convert string of length to long
+ */
+gboolean
+rspamd_strtol (const gchar *s, gsize len, glong *value)
+{
+	const gchar                    *p = s, *end = s + len;
+	gchar							c;
+	glong							v = 0;
+	const glong						cutoff = G_MAXLONG / 10, cutlim = G_MAXLONG % 10;
+	gboolean                        neg;
+
+	/* Case negative values */
+	if (*p == '-') {
+		neg = TRUE;
+		p ++;
+	}
+	else {
+		neg = FALSE;
+	}
+	/* Some preparations for range errors */
+
+	while (p < end) {
+		c = *p;
+		if (c >= '0' && c <= '9') {
+			c -= '0';
+			if (v > cutoff || (v == cutoff && c > cutlim)) {
+				/* Range error */
+				*value = neg ? G_MINLONG : G_MAXLONG;
+				return FALSE;
+			}
+			else {
+				v *= 10;
+				v += c;
+			}
+		}
+		else {
+			return FALSE;
+		}
+		p ++;
+	}
+
+	*value = neg ? -(v) : v;
+	return TRUE;
+}
+
+/*
+ * Try to convert string of length to long
+ */
+gboolean
+rspamd_strtoul (const gchar *s, gsize len, gulong *value)
+{
+	const gchar                    *p = s, *end = s + len;
+	gchar							c;
+	gulong							v = 0;
+	const gulong					cutoff = G_MAXULONG / 10, cutlim = G_MAXULONG % 10;
+
+	/* Some preparations for range errors */
+	while (p < end) {
+		c = *p;
+		if (c >= '0' && c <= '9') {
+			c -= '0';
+			if (v > cutoff || (v == cutoff && (guint8)c > cutlim)) {
+				/* Range error */
+				*value = G_MAXULONG;
+				return FALSE;
+			}
+			else {
+				v *= 10;
+				v += c;
+			}
+		}
+		else {
+			return FALSE;
+		}
+		p ++;
+	}
+
+	*value = v;
+	return TRUE;
+}
+
+/*
  * vi:ts=4
  */
diff --git a/src/util.h b/src/util.h
index a3db2d642..9d3d82416 100644
--- a/src/util.h
+++ b/src/util.h
@@ -248,4 +248,14 @@ void free_task_soft (gpointer ud);
  */
 gchar* rspamd_strncasestr (const gchar *s, const gchar *find, gint len);
 
+/*
+ * Try to convert string of length to long
+ */
+gboolean rspamd_strtol (const gchar *s, gsize len, glong *value);
+
+/*
+ * Try to convert string of length to unsigned long
+ */
+gboolean rspamd_strtoul (const gchar *s, gsize len, gulong *value);
+
 #endif