X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fpsycstore%2Fplugin_psycstore_sqlite.c;h=69291b4406338b5588c19dcbb21507084dedb31a;hb=b14f2e69adb39f4eaaaf3e26a847f29d256a1c68;hp=750fdce16f4a4882b79e5b20b39e78b993d8e806;hpb=82cf1015e43dbf2884f4abc2ef3b1bb898311517;p=oweals%2Fgnunet.git diff --git a/src/psycstore/plugin_psycstore_sqlite.c b/src/psycstore/plugin_psycstore_sqlite.c index 750fdce16..69291b440 100644 --- a/src/psycstore/plugin_psycstore_sqlite.c +++ b/src/psycstore/plugin_psycstore_sqlite.c @@ -1,22 +1,22 @@ - /* - * This file is part of GNUnet - * (C) 2009-2013 Christian Grothoff (and other contributing authors) - * - * GNUnet is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published - * by the Free Software Foundation; either version 3, or (at your - * option) any later version. - * - * GNUnet is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of -t * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with GNUnet; see the file COPYING. If not, write to the - * Free Software Foundation, Inc., 59 Temple Place - Suite 330, - * Boston, MA 02111-1307, USA. - */ +/* + * This file is part of GNUnet + * Copyright (C) 2013 GNUnet e.V. + * + * GNUnet is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published + * by the Free Software Foundation; either version 3, or (at your + * option) any later version. + * + * GNUnet is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with GNUnet; see the file COPYING. If not, write to the + * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + * Boston, MA 02110-1301, USA. + */ /** * @file psycstore/plugin_psycstore_sqlite.c @@ -35,6 +35,7 @@ t * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU #include "gnunet_psycstore_service.h" #include "gnunet_multicast_service.h" #include "gnunet_crypto_lib.h" +#include "gnunet_psyc_util_lib.h" #include "psycstore.h" #include @@ -61,6 +62,11 @@ t * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-sqlite", __VA_ARGS__) +enum Transactions { + TRANSACTION_NONE = 0, + TRANSACTION_STATE_MODIFY, + TRANSACTION_STATE_SYNC, +}; /** * Context for all functions in this plugin. @@ -80,6 +86,17 @@ struct Plugin */ sqlite3 *dbh; + /** + * Current transaction. + */ + enum Transactions transaction; + + sqlite3_stmt *transaction_begin; + + sqlite3_stmt *transaction_commit; + + sqlite3_stmt *transaction_rollback; + /** * Precompiled SQL for channel_key_store() */ @@ -115,12 +132,22 @@ struct Plugin /** * Precompiled SQL for fragment_get() */ - sqlite3_stmt *select_fragment; + sqlite3_stmt *select_fragments; + + /** + * Precompiled SQL for fragment_get() + */ + sqlite3_stmt *select_latest_fragments; + + /** + * Precompiled SQL for message_get() + */ + sqlite3_stmt *select_messages; /** * Precompiled SQL for message_get() */ - sqlite3_stmt *select_message; + sqlite3_stmt *select_latest_messages; /** * Precompiled SQL for message_get_fragment() @@ -128,21 +155,35 @@ struct Plugin sqlite3_stmt *select_message_fragment; /** - * Precompiled SQL for counters_get_master() + * Precompiled SQL for counters_get_message() */ - sqlite3_stmt *select_counters_master; + sqlite3_stmt *select_counters_message; /** - * Precompiled SQL for counters_get_slave() + * Precompiled SQL for counters_get_state() */ - sqlite3_stmt *select_counters_slave; + sqlite3_stmt *select_counters_state; + /** + * Precompiled SQL for state_modify_end() + */ + sqlite3_stmt *update_state_hash_message_id; + + /** + * Precompiled SQL for state_sync_end() + */ + sqlite3_stmt *update_max_state_message_id; /** - * Precompiled SQL for state_set() + * Precompiled SQL for state_modify_op() */ sqlite3_stmt *insert_state_current; + /** + * Precompiled SQL for state_modify_end() + */ + sqlite3_stmt *delete_state_empty; + /** * Precompiled SQL for state_set_signed() */ @@ -179,7 +220,7 @@ struct Plugin sqlite3_stmt *select_state_one; /** - * Precompiled SQL for state_get_all() + * Precompiled SQL for state_get_prefix() */ sqlite3_stmt *select_state_prefix; @@ -211,7 +252,7 @@ sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt) result = sqlite3_prepare_v2 (dbh, sql, strlen (sql), stmt, (const char **) &tail); - LOG (GNUNET_ERROR_TYPE_DEBUG, + LOG (GNUNET_ERROR_TYPE_DEBUG, "Prepared `%s' / %p: %d\n", sql, *stmt, result); if (result != SQLITE_OK) LOG (GNUNET_ERROR_TYPE_ERROR, @@ -234,7 +275,7 @@ sql_exec (sqlite3 *dbh, const char *sql) int result; result = sqlite3_exec (dbh, sql, NULL, NULL, NULL); - LOG (GNUNET_ERROR_TYPE_DEBUG, + LOG (GNUNET_ERROR_TYPE_DEBUG, "Executed `%s' / %d\n", sql, result); if (result != SQLITE_OK) LOG (GNUNET_ERROR_TYPE_ERROR, @@ -295,8 +336,9 @@ database_setup (struct Plugin *plugin) sql_exec (plugin->dbh, "PRAGMA legacy_file_format=OFF"); sql_exec (plugin->dbh, "PRAGMA auto_vacuum=INCREMENTAL"); sql_exec (plugin->dbh, "PRAGMA encoding=\"UTF-8\""); +#if ! DEBUG_PSYCSTORE sql_exec (plugin->dbh, "PRAGMA locking_mode=EXCLUSIVE"); - sql_exec (plugin->dbh, "PRAGMA count_changes=OFF"); +#endif sql_exec (plugin->dbh, "PRAGMA page_size=4096"); sqlite3_busy_timeout (plugin->dbh, BUSY_TIMEOUT_MS); @@ -306,13 +348,15 @@ database_setup (struct Plugin *plugin) sql_exec (plugin->dbh, "CREATE TABLE IF NOT EXISTS channels (\n" " id INTEGER PRIMARY KEY,\n" - " pub_key BLOB UNIQUE\n" + " pub_key BLOB(32) UNIQUE,\n" + " max_state_message_id INTEGER,\n" // last applied state message ID + " state_hash_message_id INTEGER\n" // last message ID with a state hash ");"); sql_exec (plugin->dbh, "CREATE TABLE IF NOT EXISTS slaves (\n" " id INTEGER PRIMARY KEY,\n" - " pub_key BLOB UNIQUE\n" + " pub_key BLOB(32) UNIQUE\n" ");"); sql_exec (plugin->dbh, @@ -328,6 +372,7 @@ database_setup (struct Plugin *plugin) "CREATE INDEX IF NOT EXISTS idx_membership_channel_id_slave_id " "ON membership (channel_id, slave_id);"); + /** @todo messages table: add method_name column */ sql_exec (plugin->dbh, "CREATE TABLE IF NOT EXISTS messages (\n" " channel_id INTEGER NOT NULL REFERENCES channels(id),\n" @@ -364,6 +409,12 @@ database_setup (struct Plugin *plugin) /* Prepare statements */ + sql_prepare (plugin->dbh, "BEGIN;", &plugin->transaction_begin); + + sql_prepare (plugin->dbh, "COMMIT;", &plugin->transaction_commit); + + sql_prepare (plugin->dbh, "ROLLBACK;", &plugin->transaction_rollback); + sql_prepare (plugin->dbh, "INSERT OR IGNORE INTO channels (pub_key) VALUES (?);", &plugin->insert_channel_key); @@ -390,7 +441,7 @@ database_setup (struct Plugin *plugin) &plugin->select_membership); sql_prepare (plugin->dbh, - "INSERT INTO messages\n" + "INSERT OR IGNORE INTO messages\n" " (channel_id, hop_counter, signature, purpose,\n" " fragment_id, fragment_offset, message_id,\n" " group_generation, multicast_flags, psycstore_flags, data)\n" @@ -411,17 +462,48 @@ database_setup (struct Plugin *plugin) " multicast_flags, psycstore_flags, data\n" "FROM messages\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" - " AND fragment_id = ?;", - &plugin->select_fragment); + " AND ? <= fragment_id AND fragment_id <= ?;", + &plugin->select_fragments); + /** @todo select_messages: add method_prefix filter */ sql_prepare (plugin->dbh, "SELECT hop_counter, signature, purpose, fragment_id,\n" " fragment_offset, message_id, group_generation,\n" " multicast_flags, psycstore_flags, data\n" "FROM messages\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" - " AND message_id = ?;", - &plugin->select_message); + " AND ? <= message_id AND message_id <= ?" + "LIMIT ?;", + &plugin->select_messages); + + sql_prepare (plugin->dbh, + "SELECT * FROM\n" + "(SELECT hop_counter, signature, purpose, fragment_id,\n" + " fragment_offset, message_id, group_generation,\n" + " multicast_flags, psycstore_flags, data\n" + " FROM messages\n" + " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" + " ORDER BY fragment_id DESC\n" + " LIMIT ?)\n" + "ORDER BY fragment_id;", + &plugin->select_latest_fragments); + + /** @todo select_latest_messages: add method_prefix filter */ + sql_prepare (plugin->dbh, + "SELECT hop_counter, signature, purpose, fragment_id,\n" + " fragment_offset, message_id, group_generation,\n" + " multicast_flags, psycstore_flags, data\n" + "FROM messages\n" + "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" + " AND message_id IN\n" + " (SELECT message_id\n" + " FROM messages\n" + " WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" + " GROUP BY message_id\n" + " ORDER BY message_id\n" + " DESC LIMIT ?)\n" + "ORDER BY fragment_id;", + &plugin->select_latest_messages); sql_prepare (plugin->dbh, "SELECT hop_counter, signature, purpose, fragment_id,\n" @@ -437,15 +519,25 @@ database_setup (struct Plugin *plugin) "FROM messages\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" "ORDER BY fragment_id DESC LIMIT 1;", - &plugin->select_counters_master); + &plugin->select_counters_message); sql_prepare (plugin->dbh, - "SELECT message_id\n" - "FROM messages\n" - "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" - " AND psycstore_flags & ?\n" - "ORDER BY message_id DESC LIMIT 1", - &plugin->select_counters_slave); + "SELECT max_state_message_id\n" + "FROM channels\n" + "WHERE pub_key = ? AND max_state_message_id IS NOT NULL;", + &plugin->select_counters_state); + + sql_prepare (plugin->dbh, + "UPDATE channels\n" + "SET max_state_message_id = ?\n" + "WHERE pub_key = ?;", + &plugin->update_max_state_message_id); + + sql_prepare (plugin->dbh, + "UPDATE channels\n" + "SET state_hash_message_id = ?\n" + "WHERE pub_key = ?;", + &plugin->update_state_hash_message_id); sql_prepare (plugin->dbh, "INSERT OR REPLACE INTO state\n" @@ -460,22 +552,29 @@ database_setup (struct Plugin *plugin) "ON new.channel_id = old.channel_id AND new.name = old.name;", &plugin->insert_state_current); + sql_prepare (plugin->dbh, + "DELETE FROM state\n" + "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" + " AND (value_current IS NULL OR length(value_current) = 0)\n" + " AND (value_signed IS NULL OR length(value_signed) = 0);", + &plugin->delete_state_empty); + sql_prepare (plugin->dbh, "UPDATE state\n" "SET value_signed = value_current\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);", &plugin->update_state_signed); - sql_prepare (plugin->dbh, - "INSERT INTO state_sync (channel_id, name, value)\n" - "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);", - &plugin->insert_state_sync); - sql_prepare (plugin->dbh, "DELETE FROM state\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?);", &plugin->delete_state); + sql_prepare (plugin->dbh, + "INSERT INTO state_sync (channel_id, name, value)\n" + "VALUES ((SELECT id FROM channels WHERE pub_key = ?), ?, ?);", + &plugin->insert_state_sync); + sql_prepare (plugin->dbh, "INSERT INTO state\n" " (channel_id, name, value_current, value_signed)\n" @@ -500,7 +599,7 @@ database_setup (struct Plugin *plugin) "SELECT name, value_current\n" "FROM state\n" "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n" - " AND name = ? OR name LIKE ?;", + " AND (name = ? OR substr(name, 1, ?) = ?);", &plugin->select_state_prefix); sql_prepare (plugin->dbh, @@ -524,92 +623,131 @@ database_shutdown (struct Plugin *plugin) { int result; sqlite3_stmt *stmt; + while (NULL != (stmt = sqlite3_next_stmt (plugin->dbh, NULL))) + { + result = sqlite3_finalize (stmt); + if (SQLITE_OK != result) + LOG (GNUNET_ERROR_TYPE_WARNING, + "Failed to close statement %p: %d\n", stmt, result); + } + if (SQLITE_OK != sqlite3_close (plugin->dbh)) + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close"); - if (NULL != plugin->insert_channel_key) - sqlite3_finalize (plugin->insert_channel_key); - - if (NULL != plugin->insert_slave_key) - sqlite3_finalize (plugin->insert_slave_key); - - if (NULL != plugin->insert_membership) - sqlite3_finalize (plugin->insert_membership); - - if (NULL != plugin->select_membership) - sqlite3_finalize (plugin->select_membership); - - if (NULL != plugin->insert_fragment) - sqlite3_finalize (plugin->insert_fragment); - - if (NULL != plugin->update_message_flags) - sqlite3_finalize (plugin->update_message_flags); - - if (NULL != plugin->select_fragment) - sqlite3_finalize (plugin->select_fragment); + GNUNET_free_non_null (plugin->fn); +} - if (NULL != plugin->select_message) - sqlite3_finalize (plugin->select_message); +/** + * Execute a prepared statement with a @a channel_key argument. + * + * @param plugin Plugin handle. + * @param stmt Statement to execute. + * @param channel_key Public key of the channel. + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +exec_channel (struct Plugin *plugin, sqlite3_stmt *stmt, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key) +{ + if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, + sizeof (*channel_key), SQLITE_STATIC)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_bind"); + } + else if (SQLITE_DONE != sqlite3_step (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_step"); + } - if (NULL != plugin->select_message_fragment) - sqlite3_finalize (plugin->select_message_fragment); + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + return GNUNET_SYSERR; + } - if (NULL != plugin->select_counters_master) - sqlite3_finalize (plugin->select_counters_master); + return GNUNET_OK; +} - if (NULL != plugin->select_counters_slave) - sqlite3_finalize (plugin->select_counters_slave); +/** + * Begin a transaction. + */ +static int +transaction_begin (struct Plugin *plugin, enum Transactions transaction) +{ + sqlite3_stmt *stmt = plugin->transaction_begin; - if (NULL != plugin->insert_state_current) - sqlite3_finalize (plugin->insert_state_current); + if (SQLITE_DONE != sqlite3_step (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_step"); + } + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + return GNUNET_SYSERR; + } - if (NULL != plugin->update_state_signed) - sqlite3_finalize (plugin->update_state_signed); + plugin->transaction = transaction; + return GNUNET_OK; +} - if (NULL != plugin->insert_state_sync) - sqlite3_finalize (plugin->insert_state_sync); - if (NULL != plugin->delete_state) - sqlite3_finalize (plugin->delete_state); +/** + * Commit current transaction. + */ +static int +transaction_commit (struct Plugin *plugin) +{ + sqlite3_stmt *stmt = plugin->transaction_commit; - if (NULL != plugin->insert_state_from_sync) - sqlite3_finalize (plugin->insert_state_from_sync); + if (SQLITE_DONE != sqlite3_step (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_step"); + } + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + return GNUNET_SYSERR; + } - if (NULL != plugin->delete_state_sync) - sqlite3_finalize (plugin->delete_state_sync); + plugin->transaction = TRANSACTION_NONE; + return GNUNET_OK; +} - if (NULL != plugin->select_state_one) - sqlite3_finalize (plugin->select_state_one); - if (NULL != plugin->select_state_prefix) - sqlite3_finalize (plugin->select_state_prefix); +/** + * Roll back current transaction. + */ +static int +transaction_rollback (struct Plugin *plugin) +{ + sqlite3_stmt *stmt = plugin->transaction_rollback; - result = sqlite3_close (plugin->dbh); - if (result == SQLITE_BUSY) + if (SQLITE_DONE != sqlite3_step (stmt)) { - LOG (GNUNET_ERROR_TYPE_WARNING, - _("Tried to close sqlite without finalizing all prepared statements.\n")); - stmt = sqlite3_next_stmt (plugin->dbh, NULL); - while (stmt != NULL) - { - GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "sqlite", - "Closing statement %p\n", stmt); - result = sqlite3_finalize (stmt); - if (result != SQLITE_OK) - GNUNET_log_from (GNUNET_ERROR_TYPE_WARNING, "sqlite", - "Failed to close statement %p: %d\n", stmt, result); - stmt = sqlite3_next_stmt (plugin->dbh, NULL); - } - result = sqlite3_close (plugin->dbh); + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_step"); } - if (SQLITE_OK != result) - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close"); - - GNUNET_free_non_null (plugin->fn); + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + return GNUNET_SYSERR; + } + plugin->transaction = TRANSACTION_NONE; + return GNUNET_OK; } static int channel_key_store (struct Plugin *plugin, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key) + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key) { sqlite3_stmt *stmt = plugin->insert_channel_key; @@ -617,18 +755,18 @@ channel_key_store (struct Plugin *plugin, sizeof (*channel_key), SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_channel_key (bind)"); + "sqlite3_bind"); } else if (SQLITE_DONE != sqlite3_step (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_channel_key (step)"); + "sqlite3_step"); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_channel_key (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } @@ -638,7 +776,7 @@ channel_key_store (struct Plugin *plugin, static int slave_key_store (struct Plugin *plugin, - const struct GNUNET_CRYPTO_EccPublicKey *slave_key) + const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key) { sqlite3_stmt *stmt = plugin->insert_slave_key; @@ -646,19 +784,19 @@ slave_key_store (struct Plugin *plugin, sizeof (*slave_key), SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_slave_key (bind)"); + "sqlite3_bind"); } else if (SQLITE_DONE != sqlite3_step (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_slave_key (step)"); + "sqlite3_step"); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_slave_key (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } @@ -666,7 +804,7 @@ slave_key_store (struct Plugin *plugin, } -/** +/** * Store join/leave events for a PSYC channel in order to be able to answer * membership test queries later. * @@ -675,14 +813,19 @@ slave_key_store (struct Plugin *plugin, * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -membership_store (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - const struct GNUNET_CRYPTO_EccPublicKey *slave_key, - int did_join, - uint64_t announced_at, - uint64_t effective_since, - uint64_t group_generation) +sqlite_membership_store (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key, + int did_join, + uint64_t announced_at, + uint64_t effective_since, + uint64_t group_generation) { + struct Plugin *plugin = cls; + sqlite3_stmt *stmt = plugin->insert_membership; + + GNUNET_assert (TRANSACTION_NONE == plugin->transaction); + if (announced_at > INT64_MAX || effective_since > INT64_MAX || group_generation > INT64_MAX) @@ -691,53 +834,50 @@ membership_store (void *cls, return GNUNET_SYSERR; } - struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->insert_membership; - - if (GNUNET_OK != channel_key_store (plugin, channel_key) || - GNUNET_OK != slave_key_store (plugin, slave_key)) + if (GNUNET_OK != channel_key_store (plugin, channel_key) + || GNUNET_OK != slave_key_store (plugin, slave_key)) return GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key, - sizeof (*slave_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int (stmt, 3, did_join) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 4, announced_at) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 5, effective_since) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 6, group_generation)) + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key, + sizeof (*slave_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int (stmt, 3, did_join) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 4, announced_at) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, effective_since) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, group_generation)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_membership (bind)"); + "sqlite3_bind"); } else if (SQLITE_DONE != sqlite3_step (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_membership (step)"); + "sqlite3_step"); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_membership (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } return GNUNET_OK; } -/** +/** * Test if a member was admitted to the channel at the given message ID. * * @see GNUNET_PSYCSTORE_membership_test() - * + * * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not, * #GNUNET_SYSERR if there was en error. */ static int membership_test (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - const struct GNUNET_CRYPTO_EccPublicKey *slave_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key, uint64_t message_id) { struct Plugin *plugin = cls; @@ -745,13 +885,13 @@ membership_test (void *cls, int ret = GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key, - sizeof (*slave_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id)) + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_blob (stmt, 2, slave_key, + sizeof (*slave_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_membership (bind)"); + "sqlite3_bind"); } else { @@ -768,100 +908,109 @@ membership_test (void *cls, if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_membership (reset)"); + "sqlite3_reset"); } return ret; } -/** +/** * Store a message fragment sent to a channel. * * @see GNUNET_PSYCSTORE_fragment_store() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int fragment_store (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, const struct GNUNET_MULTICAST_MessageHeader *msg, uint32_t psycstore_flags) { - if (msg->fragment_id > INT64_MAX || - msg->fragment_offset > INT64_MAX || - msg->message_id > INT64_MAX || - msg->group_generation > INT64_MAX) + struct Plugin *plugin = cls; + sqlite3_stmt *stmt = plugin->insert_fragment; + + GNUNET_assert (TRANSACTION_NONE == plugin->transaction); + + uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id); + uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset); + uint64_t message_id = GNUNET_ntohll (msg->message_id); + uint64_t group_generation = GNUNET_ntohll (msg->group_generation); + + if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX || + message_id > INT64_MAX || group_generation > INT64_MAX) { + LOG (GNUNET_ERROR_TYPE_ERROR, + "Tried to store fragment with a field > INT64_MAX: " + "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset, + message_id, group_generation); GNUNET_break (0); return GNUNET_SYSERR; } - struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->insert_fragment; - if (GNUNET_OK != channel_key_store (plugin, channel_key)) return GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 2, msg->hop_counter ) || - SQLITE_OK != sqlite3_bind_blob (stmt, 3, (const void *) &msg->signature, - sizeof (msg->signature), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_blob (stmt, 4, (const void *) &msg->purpose, - sizeof (msg->purpose), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 5, msg->fragment_id) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 6, msg->fragment_offset) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 7, msg->message_id) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 8, msg->group_generation) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 9, msg->flags) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 10, psycstore_flags) || - SQLITE_OK != sqlite3_bind_blob (stmt, 11, (const void *) &msg[1], - ntohs (msg->header.size) - sizeof (*msg), - SQLITE_STATIC)) + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, ntohl (msg->hop_counter) ) + || SQLITE_OK != sqlite3_bind_blob (stmt, 3, (const void *) &msg->signature, + sizeof (msg->signature), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_blob (stmt, 4, (const void *) &msg->purpose, + sizeof (msg->purpose), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 5, fragment_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 6, fragment_offset) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 7, message_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 8, group_generation) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 9, ntohl (msg->flags)) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 10, psycstore_flags) + || SQLITE_OK != sqlite3_bind_blob (stmt, 11, (const void *) &msg[1], + ntohs (msg->header.size) + - sizeof (*msg), SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_fragment (bind)"); + "sqlite3_bind"); } else if (SQLITE_DONE != sqlite3_step (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_fragment (step)"); + "sqlite3_step"); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_fragment (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } return GNUNET_OK; } -/** +/** * Set additional flags for a given message. * * They are OR'd with any existing flags set. - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int message_add_flags (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, uint64_t message_id, - uint64_t psycstore_flags) + uint32_t psycstore_flags) { struct Plugin *plugin = cls; sqlite3_stmt *stmt = plugin->update_message_flags; int ret = GNUNET_SYSERR; - if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, psycstore_flags) || - SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id)) + if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, psycstore_flags) + || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key, + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_id)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_message_flags (bind)"); + "sqlite3_bind"); } else { @@ -872,14 +1021,14 @@ message_add_flags (void *cls, break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_message_flags (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_message_flags (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } @@ -896,164 +1045,269 @@ fragment_row (sqlite3_stmt *stmt, GNUNET_PSYCSTORE_FragmentCallback cb, msg->header.size = htons (sizeof (*msg) + data_size); msg->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE); - msg->hop_counter = (uint32_t) sqlite3_column_int64 (stmt, 0); - memcpy (&msg->signature, + msg->hop_counter = htonl ((uint32_t) sqlite3_column_int64 (stmt, 0)); + GNUNET_memcpy (&msg->signature, sqlite3_column_blob (stmt, 1), sqlite3_column_bytes (stmt, 1)); - memcpy (&msg->purpose, + GNUNET_memcpy (&msg->purpose, sqlite3_column_blob (stmt, 2), sqlite3_column_bytes (stmt, 2)); - msg->fragment_id = sqlite3_column_int64 (stmt, 3); - msg->fragment_offset = sqlite3_column_int64 (stmt, 4); - msg->message_id = sqlite3_column_int64 (stmt, 5); - msg->group_generation = sqlite3_column_int64 (stmt, 6); - msg->flags = sqlite3_column_int64 (stmt, 7); - memcpy (&msg[1], sqlite3_column_blob (stmt, 9), data_size); + msg->fragment_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 3)); + msg->fragment_offset = GNUNET_htonll (sqlite3_column_int64 (stmt, 4)); + msg->message_id = GNUNET_htonll (sqlite3_column_int64 (stmt, 5)); + msg->group_generation = GNUNET_htonll (sqlite3_column_int64 (stmt, 6)); + msg->flags = htonl (sqlite3_column_int64 (stmt, 7)); + GNUNET_memcpy (&msg[1], sqlite3_column_blob (stmt, 9), data_size); return cb (cb_cls, (void *) msg, sqlite3_column_int64 (stmt, 8)); } -/** - * Retrieve a message fragment by fragment ID. + +static int +fragment_select (struct Plugin *plugin, sqlite3_stmt *stmt, + uint64_t *returned_fragments, + GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls) +{ + int ret = GNUNET_SYSERR; + int sql_ret; + + do + { + sql_ret = sqlite3_step (stmt); + switch (sql_ret) + { + case SQLITE_DONE: + if (ret != GNUNET_OK) + ret = GNUNET_NO; + break; + case SQLITE_ROW: + ret = fragment_row (stmt, cb, cb_cls); + (*returned_fragments)++; + if (ret != GNUNET_YES) + sql_ret = SQLITE_DONE; + break; + default: + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_step"); + } + } + while (sql_ret == SQLITE_ROW); + + return ret; +} + +/** + * Retrieve a message fragment range by fragment ID. * * @see GNUNET_PSYCSTORE_fragment_get() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int fragment_get (void *cls, - const struct - GNUNET_CRYPTO_EccPublicKey *channel_key, - uint64_t fragment_id, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t first_fragment_id, + uint64_t last_fragment_id, + uint64_t *returned_fragments, GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls) { struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->select_fragment; + sqlite3_stmt *stmt = plugin->select_fragments; int ret = GNUNET_SYSERR; + *returned_fragments = 0; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, sizeof (*channel_key), - SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 2, fragment_id)) + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_fragment_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_fragment_id)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_fragment (bind)"); + "sqlite3_bind"); } else { - switch (sqlite3_step (stmt)) - { - case SQLITE_DONE: - ret = GNUNET_NO; - break; - case SQLITE_ROW: - ret = fragment_row (stmt, cb, cb_cls); - break; - default: - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_fragment (step)"); - } + ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_fragment (reset)"); + "sqlite3_reset"); } return ret; } -/** - * Retrieve all fragments of a message. + +/** + * Retrieve a message fragment range by fragment ID. + * + * @see GNUNET_PSYCSTORE_fragment_get_latest() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +fragment_get_latest (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t fragment_limit, + uint64_t *returned_fragments, + GNUNET_PSYCSTORE_FragmentCallback cb, + void *cb_cls) +{ + struct Plugin *plugin = cls; + sqlite3_stmt *stmt = plugin->select_latest_fragments; + int ret = GNUNET_SYSERR; + *returned_fragments = 0; + + if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, + sizeof (*channel_key), + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, fragment_limit)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_bind"); + } + else + { + ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls); + } + + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + } + + return ret; +} + + +/** + * Retrieve all fragments of a message ID range. * * @see GNUNET_PSYCSTORE_message_get() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int message_get (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - uint64_t message_id, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t first_message_id, + uint64_t last_message_id, + uint64_t fragment_limit, + uint64_t *returned_fragments, GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls) { struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->select_message; + sqlite3_stmt *stmt = plugin->select_messages; int ret = GNUNET_SYSERR; + *returned_fragments = 0; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, sizeof (*channel_key), - SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id)) + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, first_message_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, last_message_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 4, + (0 != fragment_limit) + ? fragment_limit + : INT64_MAX)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message (bind)"); + "sqlite3_bind"); } else { - int sql_ret; - do - { - sql_ret = sqlite3_step (stmt); - switch (sql_ret) - { - case SQLITE_DONE: - if (ret != GNUNET_OK) - ret = GNUNET_NO; - break; - case SQLITE_ROW: - ret = fragment_row (stmt, cb, cb_cls); - if (ret != GNUNET_YES) - sql_ret = SQLITE_DONE; - break; - default: - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message (step)"); - } - } - while (sql_ret == SQLITE_ROW); + ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls); + } + + if (SQLITE_OK != sqlite3_reset (stmt)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_reset"); + } + + return ret; +} + + +/** + * Retrieve all fragments of the latest messages. + * + * @see GNUNET_PSYCSTORE_message_get_latest() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +message_get_latest (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t message_limit, + uint64_t *returned_fragments, + GNUNET_PSYCSTORE_FragmentCallback cb, + void *cb_cls) +{ + struct Plugin *plugin = cls; + sqlite3_stmt *stmt = plugin->select_latest_messages; + int ret = GNUNET_SYSERR; + *returned_fragments = 0; + + if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, + sizeof (*channel_key), + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key, + sizeof (*channel_key), + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, message_limit)) + { + LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, + "sqlite3_bind"); + } + else + { + ret = fragment_select (plugin, stmt, returned_fragments, cb, cb_cls); } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message (reset)"); + "sqlite3_reset"); } return ret; } -/** + +/** * Retrieve a fragment of message specified by its message ID and fragment * offset. * * @see GNUNET_PSYCSTORE_message_get_fragment() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int message_get_fragment (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, uint64_t message_id, uint64_t fragment_offset, GNUNET_PSYCSTORE_FragmentCallback cb, void *cb_cls) { struct Plugin *plugin = cls; - int ret = GNUNET_SYSERR; - sqlite3_stmt *stmt = plugin->select_message_fragment; + int ret = GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, sizeof (*channel_key), - SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 3, fragment_offset)) + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 2, message_id) + || SQLITE_OK != sqlite3_bind_int64 (stmt, 3, fragment_offset)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message_fragment (bind)"); + "sqlite3_bind"); } else { @@ -1067,35 +1321,35 @@ message_get_fragment (void *cls, break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message_fragment (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_message_fragment (reset)"); + "sqlite3_reset"); } return ret; } -/** - * Retrieve latest values of counters for a channel master. +/** + * Retrieve the max. values of message counters for a channel. + * + * @see GNUNET_PSYCSTORE_counters_get() * - * @see GNUNET_PSYCSTORE_counters_get_master() - * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -counters_get_master (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - uint64_t *fragment_id, - uint64_t *message_id, - uint64_t *group_generation) +counters_message_get (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t *max_fragment_id, + uint64_t *max_message_id, + uint64_t *max_group_generation) { struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->select_counters_master; + sqlite3_stmt *stmt = plugin->select_counters_message; int ret = GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, @@ -1103,7 +1357,7 @@ counters_get_master (void *cls, SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_master (bind)"); + "sqlite3_bind"); } else { @@ -1113,50 +1367,48 @@ counters_get_master (void *cls, ret = GNUNET_NO; break; case SQLITE_ROW: - *fragment_id = sqlite3_column_int64 (stmt, 0); - *message_id = sqlite3_column_int64 (stmt, 1); - *group_generation = sqlite3_column_int64 (stmt, 2); + *max_fragment_id = sqlite3_column_int64 (stmt, 0); + *max_message_id = sqlite3_column_int64 (stmt, 1); + *max_group_generation = sqlite3_column_int64 (stmt, 2); ret = GNUNET_OK; break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_master (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_master (reset)"); + "sqlite3_reset"); } return ret; } -/** - * Retrieve latest values of counters for a channel slave. +/** + * Retrieve the max. values of state counters for a channel. + * + * @see GNUNET_PSYCSTORE_counters_get() * - * @see GNUNET_PSYCSTORE_counters_get_slave() - * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -counters_get_slave (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - uint64_t *max_state_msg_id) +counters_state_get (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t *max_state_message_id) { struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->select_counters_slave; + sqlite3_stmt *stmt = plugin->select_counters_state; int ret = GNUNET_SYSERR; if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, sizeof (*channel_key), - SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_int64 (stmt, 2, - GNUNET_PSYCSTORE_MESSAGE_STATE_APPLIED)) + SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_slave (bind)"); + "sqlite3_bind"); } else { @@ -1166,67 +1418,63 @@ counters_get_slave (void *cls, ret = GNUNET_NO; break; case SQLITE_ROW: - *max_state_msg_id = sqlite3_column_int64 (stmt, 0); + *max_state_message_id = sqlite3_column_int64 (stmt, 0); ret = GNUNET_OK; break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_slave (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_counters_slave (reset)"); + "sqlite3_reset"); } return ret; } -/** - * Set a state variable to the given value. + +/** + * Assign a value to a state variable. * - * @see GNUNET_PSYCSTORE_state_modify() - * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -state_set (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - const char *name, const void *value, size_t value_size) +state_assign (struct Plugin *plugin, sqlite3_stmt *stmt, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + const char *name, const void *value, size_t value_size) { - struct Plugin *plugin = cls; int ret = GNUNET_SYSERR; - sqlite3_stmt *stmt = plugin->insert_state_current; - if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_blob (stmt, 3, value, value_size, - SQLITE_STATIC)) + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_blob (stmt, 3, value, value_size, + SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_state_current (bind)"); + "sqlite3_bind"); } else { switch (sqlite3_step (stmt)) { case SQLITE_DONE: - ret = sqlite3_total_changes (plugin->dbh) > 0 ? GNUNET_OK : GNUNET_NO; + ret = 0 < sqlite3_total_changes (plugin->dbh) ? GNUNET_OK : GNUNET_NO; break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_state_current (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "insert_state_current (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } @@ -1234,88 +1482,235 @@ state_set (void *cls, } -/** - * Reset the state of a channel. - * - * @see GNUNET_PSYCSTORE_state_reset() - * - * @return #GNUNET_OK on success, else #GNUNET_SYSERR - */ static int -state_reset (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key) +update_message_id (struct Plugin *plugin, sqlite3_stmt *stmt, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t message_id) { - struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->delete_state; - - if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC)) + if (SQLITE_OK != sqlite3_bind_int64 (stmt, 1, message_id) + || SQLITE_OK != sqlite3_bind_blob (stmt, 2, channel_key, + sizeof (*channel_key), SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "delete_state (bind)"); + "sqlite3_bind"); } else if (SQLITE_DONE != sqlite3_step (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "delete_state (step)"); + "sqlite3_step"); } - if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "delete_state (reset)"); + "sqlite3_reset"); return GNUNET_SYSERR; } - return GNUNET_OK; } -/** - * Update signed values of state variables in the state store. - * - * @see GNUNET_PSYCSTORE_state_hash_update() - * - * @return #GNUNET_OK on success, else #GNUNET_SYSERR +/** + * Begin modifying current state. */ static int -state_update_signed (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key) +state_modify_begin (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t message_id, uint64_t state_delta) { struct Plugin *plugin = cls; - sqlite3_stmt *stmt = plugin->update_state_signed; - if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC)) + if (state_delta > 0) { - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_state_signed (bind)"); + /** + * We can only apply state modifiers in the current message if modifiers in + * the previous stateful message (message_id - state_delta) were already + * applied. + */ + + uint64_t max_state_message_id = 0; + int ret = counters_state_get (plugin, channel_key, &max_state_message_id); + switch (ret) + { + case GNUNET_OK: + case GNUNET_NO: // no state yet + ret = GNUNET_OK; + break; + default: + return ret; + } + + if (max_state_message_id < message_id - state_delta) + return GNUNET_NO; /* some stateful messages not yet applied */ + else if (message_id - state_delta < max_state_message_id) + return GNUNET_NO; /* changes already applied */ } - else if (SQLITE_DONE != sqlite3_step (stmt)) + + if (TRANSACTION_NONE != plugin->transaction) { - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_state_signed (step)"); + /** @todo FIXME: wait for other transaction to finish */ + return GNUNET_SYSERR; } + return transaction_begin (plugin, TRANSACTION_STATE_MODIFY); +} - if (SQLITE_OK != sqlite3_reset (stmt)) + +/** + * Set the current value of state variable. + * + * @see GNUNET_PSYCSTORE_state_modify() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +state_modify_op (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + enum GNUNET_PSYC_Operator op, + const char *name, const void *value, size_t value_size) +{ + struct Plugin *plugin = cls; + GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction); + + switch (op) { - LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "update_state_signed (reset)"); + case GNUNET_PSYC_OP_ASSIGN: + return state_assign (plugin, plugin->insert_state_current, channel_key, + name, value, value_size); + + default: /** @todo implement more state operations */ + GNUNET_break (0); return GNUNET_SYSERR; } +} - return GNUNET_OK; + +/** + * End modifying current state. + */ +static int +state_modify_end (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t message_id) +{ + struct Plugin *plugin = cls; + GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction); + + return + GNUNET_OK == exec_channel (plugin, plugin->delete_state_empty, channel_key) + && GNUNET_OK == update_message_id (plugin, + plugin->update_max_state_message_id, + channel_key, message_id) + && GNUNET_OK == transaction_commit (plugin) + ? GNUNET_OK : GNUNET_SYSERR; +} + + +/** + * Begin state synchronization. + */ +static int +state_sync_begin (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key) +{ + struct Plugin *plugin = cls; + return exec_channel (plugin, plugin->delete_state_sync, channel_key); } -/** +/** + * Assign current value of a state variable. + * + * @see GNUNET_PSYCSTORE_state_modify() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +state_sync_assign (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + const char *name, const void *value, size_t value_size) +{ + struct Plugin *plugin = cls; + return state_assign (cls, plugin->insert_state_sync, channel_key, + name, value, value_size); +} + + +/** + * End modifying current state. + */ +static int +state_sync_end (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + uint64_t max_state_message_id, + uint64_t state_hash_message_id) +{ + struct Plugin *plugin = cls; + int ret = GNUNET_SYSERR; + + if (TRANSACTION_NONE != plugin->transaction) + { + /** @todo FIXME: wait for other transaction to finish */ + return GNUNET_SYSERR; + } + + GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC) + && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key) + && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync, + channel_key) + && GNUNET_OK == exec_channel (plugin, plugin->delete_state_sync, + channel_key) + && GNUNET_OK == update_message_id (plugin, + plugin->update_state_hash_message_id, + channel_key, state_hash_message_id) + && GNUNET_OK == update_message_id (plugin, + plugin->update_max_state_message_id, + channel_key, max_state_message_id) + && GNUNET_OK == transaction_commit (plugin) + ? ret = GNUNET_OK + : transaction_rollback (plugin); + return ret; +} + + +/** + * Delete the whole state. + * + * @see GNUNET_PSYCSTORE_state_reset() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key) +{ + struct Plugin *plugin = cls; + return exec_channel (plugin, plugin->delete_state, channel_key); +} + + +/** + * Update signed values of state variables in the state store. + * + * @see GNUNET_PSYCSTORE_state_hash_update() + * + * @return #GNUNET_OK on success, else #GNUNET_SYSERR + */ +static int +state_update_signed (void *cls, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key) +{ + struct Plugin *plugin = cls; + return exec_channel (plugin, plugin->update_state_signed, channel_key); +} + + +/** * Retrieve a state variable by name. * * @see GNUNET_PSYCSTORE_state_get() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -state_get (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key, +state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls) { struct Plugin *plugin = cls; @@ -1325,11 +1720,11 @@ state_get (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key, if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, sizeof (*channel_key), - SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC)) + SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, -1, SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_one (bind)"); + "sqlite3_bind"); } else { @@ -1344,50 +1739,45 @@ state_get (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key, break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_one (step)"); + "sqlite3_step"); } } if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_one (reset)"); + "sqlite3_reset"); } - return ret; } -/** +/** * Retrieve all state variables for a channel with the given prefix. * - * @see GNUNET_PSYCSTORE_state_get_all() - * + * @see GNUNET_PSYCSTORE_state_get_prefix() + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int -state_get_all (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key, - const char *name, GNUNET_PSYCSTORE_StateCallback cb, - void *cb_cls) +state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, + const char *name, GNUNET_PSYCSTORE_StateCallback cb, + void *cb_cls) { struct Plugin *plugin = cls; int ret = GNUNET_SYSERR; - sqlite3_stmt *stmt = plugin->select_state_prefix; size_t name_len = strlen (name); - char *name_prefix = GNUNET_malloc (name_len + 2); - memcpy (name_prefix, name, name_len); - memcpy (name_prefix + name_len, "_%", 2); if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key, - sizeof (*channel_key), SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC) || - SQLITE_OK != sqlite3_bind_text (stmt, 3, name_prefix, name_len + 2, - SQLITE_STATIC)) + sizeof (*channel_key), SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC) + || SQLITE_OK != sqlite3_bind_int (stmt, 3, name_len) + || SQLITE_OK != sqlite3_bind_text (stmt, 4, name, name_len, SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_prefix (bind)"); + "sqlite3_bind"); } else { @@ -1410,32 +1800,30 @@ state_get_all (void *cls, const struct GNUNET_CRYPTO_EccPublicKey *channel_key, break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_prefix (step)"); + "sqlite3_step"); } } while (sql_ret == SQLITE_ROW); } - if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_prefix (reset)"); + "sqlite3_reset"); } - return ret; } -/** +/** * Retrieve all signed state variables for a channel. * * @see GNUNET_PSYCSTORE_state_get_signed() - * + * * @return #GNUNET_OK on success, else #GNUNET_SYSERR */ static int state_get_signed (void *cls, - const struct GNUNET_CRYPTO_EccPublicKey *channel_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls) { struct Plugin *plugin = cls; @@ -1447,7 +1835,7 @@ state_get_signed (void *cls, sizeof (*channel_key), SQLITE_STATIC)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_signed (bind)"); + "sqlite3_bind"); } else { @@ -1470,7 +1858,7 @@ state_get_signed (void *cls, break; default: LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_signed (step)"); + "sqlite3_step"); } } while (sql_ret == SQLITE_ROW); @@ -1479,14 +1867,14 @@ state_get_signed (void *cls, if (SQLITE_OK != sqlite3_reset (stmt)) { LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, - "select_state_signed (reset)"); + "sqlite3_reset"); } return ret; } -/** +/** * Entry point for the plugin. * * @param cls The struct GNUNET_CONFIGURATION_Handle. @@ -1502,7 +1890,7 @@ libgnunet_plugin_psycstore_sqlite_init (void *cls) if (NULL != plugin.cfg) return NULL; /* can only initialize once! */ memset (&plugin, 0, sizeof (struct Plugin)); - plugin.cfg = cfg; + plugin.cfg = cfg; if (GNUNET_OK != database_setup (&plugin)) { database_shutdown (&plugin); @@ -1510,20 +1898,27 @@ libgnunet_plugin_psycstore_sqlite_init (void *cls) } api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions); api->cls = &plugin; - api->membership_store = &membership_store; + api->membership_store = &sqlite_membership_store; api->membership_test = &membership_test; api->fragment_store = &fragment_store; api->message_add_flags = &message_add_flags; api->fragment_get = &fragment_get; + api->fragment_get_latest = &fragment_get_latest; api->message_get = &message_get; + api->message_get_latest = &message_get_latest; api->message_get_fragment = &message_get_fragment; - api->counters_get_master = &counters_get_master; - api->counters_get_slave = &counters_get_slave; - api->state_set = &state_set; + api->counters_message_get = &counters_message_get; + api->counters_state_get = &counters_state_get; + api->state_modify_begin = &state_modify_begin; + api->state_modify_op = &state_modify_op; + api->state_modify_end = &state_modify_end; + api->state_sync_begin = &state_sync_begin; + api->state_sync_assign = &state_sync_assign; + api->state_sync_end = &state_sync_end; api->state_reset = &state_reset; api->state_update_signed = &state_update_signed; api->state_get = &state_get; - api->state_get_all = &state_get_all; + api->state_get_prefix = &state_get_prefix; api->state_get_signed = &state_get_signed; LOG (GNUNET_ERROR_TYPE_INFO, _("SQLite database running\n"));