plugin_datastore_sqlite.c

Go to the documentation of this file.

/*
 * This file is part of GNUnet
 * Copyright (C) 2009, 2011, 2017 GNUnet e.V.
 *
 * GNUnet is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Affero General Public License as published
 * by the Free Software Foundation, either version 3 of the License,
 * 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
 * Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.

     SPDX-License-Identifier: AGPL3.0-or-later
 */

#include "platform.h"
#include "gnunet_datastore_plugin.h"
#include "gnunet_sq_lib.h"
#include <sqlite3.h>


#define MAX_ITEM_SIZE 65536

#define BUSY_TIMEOUT_MS 250


#define LOG_SQLITE(db, level, cmd)                                \
  do                                                              \
    {                                                               \
      GNUNET_log_from(level,                                       \
                      "sqlite",                                    \
                      _("`%s' failed at %s:%d with error: %s\n"), \
                      cmd,                                         \
                      __FILE__,                                    \
                      __LINE__,                                    \
                      sqlite3_errmsg(db->dbh));                   \
    } while (0)


#define LOG_SQLITE_MSG(db, msg, level, cmd)                       \
  do                                                              \
    {                                                               \
      GNUNET_log_from(level,                                       \
                      "sqlite",                                    \
                      _("`%s' failed at %s:%d with error: %s\n"), \
                      cmd,                                         \
                      __FILE__,                                    \
                      __LINE__,                                    \
                      sqlite3_errmsg(db->dbh));                   \
      GNUNET_asprintf(msg,                                         \
                      _("`%s' failed at %s:%u with error: %s"),   \
                      cmd,                                         \
                      __FILE__,                                    \
                      __LINE__,                                    \
                      sqlite3_errmsg(db->dbh));                   \
    } while (0)


struct Plugin {
  struct GNUNET_DATASTORE_PluginEnvironment *env;

  char *fn;

  sqlite3 *dbh;

  sqlite3_stmt *remove;

  sqlite3_stmt *delRow;

  sqlite3_stmt *update;

  sqlite3_stmt *maxRepl;

  sqlite3_stmt *updRepl;

  sqlite3_stmt *selRepl;

  sqlite3_stmt *selExpi;

  sqlite3_stmt *selZeroAnon;

  sqlite3_stmt *insertContent;

  sqlite3_stmt *get[8];

  int drop_on_shutdown;
};


static int
sq_prepare(sqlite3 *dbh, const char *zSql, sqlite3_stmt **ppStmt)
{
  char *dummy;
  int result;

  result = sqlite3_prepare_v2(dbh,
                              zSql,
                              strlen(zSql),
                              ppStmt,
                              (const char **)&dummy);
  GNUNET_log_from(GNUNET_ERROR_TYPE_DEBUG,
                  "sqlite",
                  "Prepared `%s' / %p: %d\n",
                  zSql,
                  *ppStmt,
                  result);
  return result;
}


static void
create_indices(sqlite3 *dbh)
{
  /* create indices */
  if (
    0 !=
    (SQLITE_OK !=
     sqlite3_exec(dbh,
                  "CREATE INDEX IF NOT EXISTS idx_hash ON gn091 (hash)",
                  NULL,
                  NULL,
                  NULL)) +
    (SQLITE_OK !=
     sqlite3_exec(
       dbh,
       "CREATE INDEX IF NOT EXISTS idx_anon_type ON gn091 (anonLevel ASC,type)",
       NULL,
       NULL,
       NULL)) +
    (SQLITE_OK !=
     sqlite3_exec(dbh,
                  "CREATE INDEX IF NOT EXISTS idx_expire ON gn091 (expire ASC)",
                  NULL,
                  NULL,
                  NULL)) +
    (SQLITE_OK !=
     sqlite3_exec(
       dbh,
       "CREATE INDEX IF NOT EXISTS idx_repl_rvalue ON gn091 (repl,rvalue)",
       NULL,
       NULL,
       NULL)))
    GNUNET_log_from(GNUNET_ERROR_TYPE_ERROR,
                    "sqlite",
                    "Failed to create indices: %s\n",
                    sqlite3_errmsg(dbh));
}


#if 0
#define CHECK(a) GNUNET_break(a)
#define ENULL NULL
#else
#define ENULL &e
#define ENULL_DEFINED 1
#define CHECK(a)                                     \
  if (!(a))                                         \
    {                                                  \
      GNUNET_log(GNUNET_ERROR_TYPE_ERROR, "%s\n", e); \
      sqlite3_free(e);                                \
    }
#endif


static int
database_setup(const struct GNUNET_CONFIGURATION_Handle *cfg,
               struct Plugin *plugin)
{
  sqlite3_stmt *stmt;
  char *afsdir;

#if ENULL_DEFINED
  char *e;
#endif

  if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_filename(cfg,
                                                           "datastore-sqlite",
                                                           "FILENAME",
                                                           &afsdir))
    {
      GNUNET_log_config_missing(GNUNET_ERROR_TYPE_ERROR,
                                "datastore-sqlite",
                                "FILENAME");
      return GNUNET_SYSERR;
    }
  if (GNUNET_OK != GNUNET_DISK_file_test(afsdir))
    {
      if (GNUNET_OK != GNUNET_DISK_directory_create_for_file(afsdir))
        {
          GNUNET_break(0);
          GNUNET_free(afsdir);
          return GNUNET_SYSERR;
        }
      /* database is new or got deleted, reset payload to zero! */
      if (NULL != plugin->env->duc)
        plugin->env->duc(plugin->env->cls, 0);
    }
  /* afsdir should be UTF-8-encoded. If it isn't, it's a bug */
  plugin->fn = afsdir;

  /* Open database and precompile statements */
  if (SQLITE_OK != sqlite3_open(plugin->fn, &plugin->dbh))
    {
      GNUNET_log_from(GNUNET_ERROR_TYPE_ERROR,
                      "sqlite",
                      _("Unable to initialize SQLite: %s.\n"),
                      sqlite3_errmsg(plugin->dbh));
      return GNUNET_SYSERR;
    }
  CHECK(
    SQLITE_OK ==
    sqlite3_exec(plugin->dbh, "PRAGMA temp_store=MEMORY", NULL, NULL, ENULL));
  CHECK(
    SQLITE_OK ==
    sqlite3_exec(plugin->dbh, "PRAGMA synchronous=OFF", NULL, NULL, ENULL));
  CHECK(SQLITE_OK == sqlite3_exec(plugin->dbh,
                                  "PRAGMA legacy_file_format=OFF",
                                  NULL,
                                  NULL,
                                  ENULL));
  CHECK(SQLITE_OK == sqlite3_exec(plugin->dbh,
                                  "PRAGMA auto_vacuum=INCREMENTAL",
                                  NULL,
                                  NULL,
                                  ENULL));
  CHECK(SQLITE_OK == sqlite3_exec(plugin->dbh,
                                  "PRAGMA locking_mode=EXCLUSIVE",
                                  NULL,
                                  NULL,
                                  ENULL));
  CHECK(
    SQLITE_OK ==
    sqlite3_exec(plugin->dbh, "PRAGMA page_size=4096", NULL, NULL, ENULL));

  CHECK(SQLITE_OK == sqlite3_busy_timeout(plugin->dbh, BUSY_TIMEOUT_MS));


  /* We have to do it here, because otherwise precompiling SQL might fail */
  CHECK(SQLITE_OK ==
        sq_prepare(plugin->dbh,
                   "SELECT 1 FROM sqlite_master WHERE tbl_name = 'gn091'",
                   &stmt));

  /* FIXME: SQLite does not have unsigned integers! This is ok for the type column because
   * we only test equality on it and can cast it to/from uint32_t. For repl, prio, and anonLevel
   * we do math or inequality tests, so we can't handle the entire range of uint32_t.
   * This will also cause problems for expiration times after 294247-01-10-04:00:54 UTC.
   */
  if ((SQLITE_DONE == sqlite3_step(stmt)) &&
      (SQLITE_OK != sqlite3_exec(plugin->dbh,
                                 "CREATE TABLE gn091 ("
                                 "  repl INT4 NOT NULL DEFAULT 0,"
                                 "  type INT4 NOT NULL DEFAULT 0,"
                                 "  prio INT4 NOT NULL DEFAULT 0,"
                                 "  anonLevel INT4 NOT NULL DEFAULT 0,"
                                 "  expire INT8 NOT NULL DEFAULT 0,"
                                 "  rvalue INT8 NOT NULL,"
                                 "  hash TEXT NOT NULL DEFAULT '',"
                                 "  vhash TEXT NOT NULL DEFAULT '',"
                                 "  value BLOB NOT NULL DEFAULT '')",
                                 NULL,
                                 NULL,
                                 NULL)))
    {
      LOG_SQLITE(plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_exec");
      sqlite3_finalize(stmt);
      return GNUNET_SYSERR;
    }
  sqlite3_finalize(stmt);
  create_indices(plugin->dbh);

#define RESULT_COLUMNS \
  "repl, type, prio, anonLevel, expire, hash, value, _ROWID_"
  if (
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "UPDATE gn091 "
                             "SET prio = prio + ?, "
                             "repl = repl + ?, "
                             "expire = MAX(expire, ?) "
                             "WHERE hash = ? AND vhash = ?",
                             &plugin->update)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "UPDATE gn091 "
                             "SET repl = MAX (0, repl - 1) WHERE _ROWID_ = ?",
                             &plugin->updRepl)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE repl=?2 AND "
                             " (rvalue>=?1 OR "
                             "  NOT EXISTS (SELECT 1 FROM gn091 "
                             "WHERE repl=?2 AND rvalue>=?1 LIMIT 1) ) "
                             "ORDER BY rvalue ASC LIMIT 1",
                             &plugin->selRepl)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT MAX(repl) FROM gn091",
                             &plugin->maxRepl)) ||
    (SQLITE_OK !=
     sq_prepare(plugin->dbh,
                "SELECT " RESULT_COLUMNS " FROM gn091 "
                "WHERE NOT EXISTS (SELECT 1 FROM gn091 WHERE expire < ?1 LIMIT 1) OR (expire < ?1) "
                "ORDER BY expire ASC LIMIT 1",
                &plugin->selExpi)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ? AND "
                             "anonLevel = 0 AND "
                             "type = ? "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->selZeroAnon)) ||
    (SQLITE_OK !=
     sq_prepare(plugin->dbh,
                "INSERT INTO gn091 (repl, type, prio, anonLevel, expire, rvalue, hash, vhash, value) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                &plugin->insertContent)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[0])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "type = ?4 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[1])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "hash = ?3 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[2])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "hash = ?3 AND "
                             "type = ?4 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[3])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "rvalue >= ?2 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[4])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "rvalue >= ?2 AND "
                             "type = ?4 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[5])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "rvalue >= ?2 AND "
                             "hash = ?3 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[6])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "SELECT " RESULT_COLUMNS " FROM gn091 "
                             "WHERE _ROWID_ >= ?1 AND "
                             "rvalue >= ?2 AND "
                             "hash = ?3 AND "
                             "type = ?4 "
                             "ORDER BY _ROWID_ ASC LIMIT 1",
                             &plugin->get[7])) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "DELETE FROM gn091 WHERE _ROWID_ = ?",
                             &plugin->delRow)) ||
    (SQLITE_OK != sq_prepare(plugin->dbh,
                             "DELETE FROM gn091 "
                             "WHERE hash = ? AND "
                             "value = ? ",
                             &plugin->remove)) ||
    false)
    {
      LOG_SQLITE(plugin, GNUNET_ERROR_TYPE_ERROR, "precompiling");
      return GNUNET_SYSERR;
    }
  return GNUNET_OK;
}


static void
database_shutdown(struct Plugin *plugin)
{
  int result;

#if SQLITE_VERSION_NUMBER >= 3007000
  sqlite3_stmt *stmt;
#endif

  if (NULL != plugin->remove)
    sqlite3_finalize(plugin->remove);
  if (NULL != plugin->delRow)
    sqlite3_finalize(plugin->delRow);
  if (NULL != plugin->update)
    sqlite3_finalize(plugin->update);
  if (NULL != plugin->updRepl)
    sqlite3_finalize(plugin->updRepl);
  if (NULL != plugin->selRepl)
    sqlite3_finalize(plugin->selRepl);
  if (NULL != plugin->maxRepl)
    sqlite3_finalize(plugin->maxRepl);
  if (NULL != plugin->selExpi)
    sqlite3_finalize(plugin->selExpi);
  if (NULL != plugin->selZeroAnon)
    sqlite3_finalize(plugin->selZeroAnon);
  if (NULL != plugin->insertContent)
    sqlite3_finalize(plugin->insertContent);
  for (int i = 0; i < 8; ++i)
    if (NULL != plugin->get[i])
      sqlite3_finalize(plugin->get[i]);
  result = sqlite3_close(plugin->dbh);
#if SQLITE_VERSION_NUMBER >= 3007000
  if (result == SQLITE_BUSY)
    {
      GNUNET_log_from(
        GNUNET_ERROR_TYPE_WARNING,
        "sqlite",
        _(
          "Tried to close sqlite without finalizing all prepared statements.\n"));
      stmt = sqlite3_next_stmt(plugin->dbh, NULL);
      while (NULL != stmt)
        {
          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);
    }
#endif
  if (SQLITE_OK != result)
    LOG_SQLITE(plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite3_close");
  GNUNET_free_non_null(plugin->fn);
}


static int
delete_by_rowid(struct Plugin *plugin, uint64_t rid)
{
  struct GNUNET_SQ_QueryParam params[] = { GNUNET_SQ_query_param_uint64(&rid),
                                           GNUNET_SQ_query_param_end };

  if (GNUNET_OK != GNUNET_SQ_bind(plugin->delRow, params))
    return GNUNET_SYSERR;
  if (SQLITE_DONE != sqlite3_step(plugin->delRow))
    {
      LOG_SQLITE(plugin,
                 GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_step");
      GNUNET_SQ_reset(plugin->dbh, plugin->delRow);
      return GNUNET_SYSERR;
    }
  GNUNET_SQ_reset(plugin->dbh, plugin->delRow);
  return GNUNET_OK;
}


static void
sqlite_plugin_put(void *cls,
                  const struct GNUNET_HashCode *key,
                  bool absent,
                  uint32_t size,
                  const void *data,
                  enum GNUNET_BLOCK_Type type,
                  uint32_t priority,
                  uint32_t anonymity,
                  uint32_t replication,
                  struct GNUNET_TIME_Absolute expiration,
                  PluginPutCont cont,
                  void *cont_cls)
{
  struct Plugin *plugin = cls;
  struct GNUNET_HashCode vhash;
  char *msg = NULL;

  GNUNET_CRYPTO_hash(data, size, &vhash);

  if (!absent)
    {
      struct GNUNET_SQ_QueryParam params[] =
      { GNUNET_SQ_query_param_uint32(&priority),
        GNUNET_SQ_query_param_uint32(&replication),
        GNUNET_SQ_query_param_absolute_time(&expiration),
        GNUNET_SQ_query_param_auto_from_type(key),
        GNUNET_SQ_query_param_auto_from_type(&vhash),
        GNUNET_SQ_query_param_end };

      if (GNUNET_OK != GNUNET_SQ_bind(plugin->update, params))
        {
          cont(cont_cls, key, size, GNUNET_SYSERR, _("sqlite bind failure"));
          return;
        }
      if (SQLITE_DONE != sqlite3_step(plugin->update))
        {
          LOG_SQLITE_MSG(plugin,
                         &msg,
                         GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                         "sqlite3_step");
          cont(cont_cls, key, size, GNUNET_SYSERR, msg);
          GNUNET_free_non_null(msg);
          return;
        }
      int changes = sqlite3_changes(plugin->dbh);
      GNUNET_SQ_reset(plugin->dbh, plugin->update);
      if (0 != changes)
        {
          cont(cont_cls, key, size, GNUNET_NO, NULL);
          return;
        }
    }

  uint64_t rvalue;
  uint32_t type32 = (uint32_t)type;
  struct GNUNET_SQ_QueryParam params[] =
  { GNUNET_SQ_query_param_uint32(&replication),
    GNUNET_SQ_query_param_uint32(&type32),
    GNUNET_SQ_query_param_uint32(&priority),
    GNUNET_SQ_query_param_uint32(&anonymity),
    GNUNET_SQ_query_param_absolute_time(&expiration),
    GNUNET_SQ_query_param_uint64(&rvalue),
    GNUNET_SQ_query_param_auto_from_type(key),
    GNUNET_SQ_query_param_auto_from_type(&vhash),
    GNUNET_SQ_query_param_fixed_size(data, size),
    GNUNET_SQ_query_param_end };
  int n;
  int ret;
  sqlite3_stmt *stmt;

  if (size > MAX_ITEM_SIZE)
    {
      cont(cont_cls, key, size, GNUNET_SYSERR, _("Data too large"));
      return;
    }
  GNUNET_log_from(
    GNUNET_ERROR_TYPE_DEBUG,
    "sqlite",
    "Storing in database block with type %u/key `%s'/priority %u/expiration in %s (%s).\n",
    type,
    GNUNET_h2s(key),
    priority,
    GNUNET_STRINGS_relative_time_to_string(GNUNET_TIME_absolute_get_remaining(
                                             expiration),
                                           GNUNET_YES),
    GNUNET_STRINGS_absolute_time_to_string(expiration));
  stmt = plugin->insertContent;
  rvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
  if (GNUNET_OK != GNUNET_SQ_bind(stmt, params))
    {
      cont(cont_cls, key, size, GNUNET_SYSERR, NULL);
      return;
    }
  n = sqlite3_step(stmt);
  switch (n)
    {
    case SQLITE_DONE:
      if (NULL != plugin->env->duc)
        plugin->env->duc(plugin->env->cls,
                         size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
      GNUNET_log_from(GNUNET_ERROR_TYPE_DEBUG,
                      "sqlite",
                      "Stored new entry (%u bytes)\n",
                      size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
      ret = GNUNET_OK;
      break;

    case SQLITE_BUSY:
      GNUNET_break(0);
      LOG_SQLITE_MSG(plugin,
                     &msg,
                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                     "sqlite3_step");
      ret = GNUNET_SYSERR;
      break;

    default:
      LOG_SQLITE_MSG(plugin,
                     &msg,
                     GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                     "sqlite3_step");
      GNUNET_SQ_reset(plugin->dbh, stmt);
      database_shutdown(plugin);
      database_setup(plugin->env->cfg, plugin);
      cont(cont_cls, key, size, GNUNET_SYSERR, msg);
      GNUNET_free_non_null(msg);
      return;
    }
  GNUNET_SQ_reset(plugin->dbh, stmt);
  cont(cont_cls, key, size, ret, msg);
  GNUNET_free_non_null(msg);
}


static void
execute_get(struct Plugin *plugin,
            sqlite3_stmt *stmt,
            PluginDatumProcessor proc,
            void *proc_cls)
{
  int n;
  struct GNUNET_TIME_Absolute expiration;
  uint32_t replication;
  uint32_t type;
  uint32_t priority;
  uint32_t anonymity;
  uint64_t rowid;
  void *value;
  size_t value_size;
  struct GNUNET_HashCode key;
  int ret;
  struct GNUNET_SQ_ResultSpec rs[] =
  { GNUNET_SQ_result_spec_uint32(&replication),
    GNUNET_SQ_result_spec_uint32(&type),
    GNUNET_SQ_result_spec_uint32(&priority),
    GNUNET_SQ_result_spec_uint32(&anonymity),
    GNUNET_SQ_result_spec_absolute_time(&expiration),
    GNUNET_SQ_result_spec_auto_from_type(&key),
    GNUNET_SQ_result_spec_variable_size(&value, &value_size),
    GNUNET_SQ_result_spec_uint64(&rowid),
    GNUNET_SQ_result_spec_end };

  n = sqlite3_step(stmt);
  switch (n)
    {
    case SQLITE_ROW:
      if (GNUNET_OK != GNUNET_SQ_extract_result(stmt, rs))
        {
          GNUNET_break(0);
          break;
        }
      GNUNET_log_from(GNUNET_ERROR_TYPE_DEBUG,
                      "sqlite",
                      "Found reply in database with expiration %s\n",
                      GNUNET_STRINGS_absolute_time_to_string(expiration));
      ret = proc(proc_cls,
                 &key,
                 value_size,
                 value,
                 type,
                 priority,
                 anonymity,
                 replication,
                 expiration,
                 rowid);
      GNUNET_SQ_cleanup_result(rs);
      GNUNET_SQ_reset(plugin->dbh, stmt);
      if ((GNUNET_NO == ret) && (GNUNET_OK == delete_by_rowid(plugin, rowid)) &&
          (NULL != plugin->env->duc))
        plugin->env->duc(plugin->env->cls,
                         -(value_size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
      return;

    case SQLITE_DONE:
      /* database must be empty */
      break;

    case SQLITE_BUSY:
    case SQLITE_ERROR:
    case SQLITE_MISUSE:
    default:
      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");
      GNUNET_break(0);
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      database_shutdown(plugin);
      database_setup(plugin->env->cfg, plugin);
      return;
    }
  GNUNET_SQ_reset(plugin->dbh, stmt);
  proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
}


static void
sqlite_plugin_get_zero_anonymity(void *cls,
                                 uint64_t next_uid,
                                 enum GNUNET_BLOCK_Type type,
                                 PluginDatumProcessor proc,
                                 void *proc_cls)
{
  struct Plugin *plugin = cls;
  uint32_t type32 = type;
  struct GNUNET_SQ_QueryParam params[] = { GNUNET_SQ_query_param_uint64(
                                             &next_uid),
                                           GNUNET_SQ_query_param_uint32(
                                             &type32),
                                           GNUNET_SQ_query_param_end };

  GNUNET_assert(type != GNUNET_BLOCK_TYPE_ANY);
  if (GNUNET_OK != GNUNET_SQ_bind(plugin->selZeroAnon, params))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  execute_get(plugin, plugin->selZeroAnon, proc, proc_cls);
}


static void
sqlite_plugin_get_key(void *cls,
                      uint64_t next_uid,
                      bool random,
                      const struct GNUNET_HashCode *key,
                      enum GNUNET_BLOCK_Type type,
                      PluginDatumProcessor proc,
                      void *proc_cls)
{
  struct Plugin *plugin = cls;
  uint64_t rvalue;
  int use_rvalue = random;
  uint32_t type32 = (uint32_t)type;
  int use_type = GNUNET_BLOCK_TYPE_ANY != type;
  int use_key = NULL != key;
  sqlite3_stmt *stmt = plugin->get[use_rvalue * 4 + use_key * 2 + use_type];
  struct GNUNET_SQ_QueryParam params[] =
  { GNUNET_SQ_query_param_uint64(&next_uid),
    GNUNET_SQ_query_param_uint64(&rvalue),
    GNUNET_SQ_query_param_auto_from_type(key),
    GNUNET_SQ_query_param_uint32(&type32),
    GNUNET_SQ_query_param_end };

  /* SQLite doesn't like it when you try to bind a parameter greater than the
   * last numbered parameter, but unused parameters in the middle are OK.
   */
  if (!use_type)
    {
      params[3] = (struct GNUNET_SQ_QueryParam)GNUNET_SQ_query_param_end;
      if (!use_key)
        {
          params[2] = (struct GNUNET_SQ_QueryParam)GNUNET_SQ_query_param_end;
          if (!use_rvalue)
            params[1] = (struct GNUNET_SQ_QueryParam)GNUNET_SQ_query_param_end;
        }
    }
  if (random)
    {
      rvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
      next_uid = 0;
    }
  else
    rvalue = 0;

  if (GNUNET_OK != GNUNET_SQ_bind(stmt, params))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  execute_get(plugin, stmt, proc, proc_cls);
}


struct ReplCtx {
  PluginDatumProcessor proc;

  void *proc_cls;

  uint64_t uid;

  int have_uid;
};


static int
repl_proc(void *cls,
          const struct GNUNET_HashCode *key,
          uint32_t size,
          const void *data,
          enum GNUNET_BLOCK_Type type,
          uint32_t priority,
          uint32_t anonymity,
          uint32_t replication,
          struct GNUNET_TIME_Absolute expiration,
          uint64_t uid)
{
  struct ReplCtx *rc = cls;
  int ret;

  if (GNUNET_SYSERR == rc->have_uid)
    rc->have_uid = GNUNET_NO;
  ret = rc->proc(rc->proc_cls,
                 key,
                 size,
                 data,
                 type,
                 priority,
                 anonymity,
                 replication,
                 expiration,
                 uid);
  if (NULL != key)
    {
      rc->uid = uid;
      rc->have_uid = GNUNET_YES;
    }
  return ret;
}


static void
sqlite_plugin_get_replication(void *cls,
                              PluginDatumProcessor proc,
                              void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct ReplCtx rc;
  uint64_t rvalue;
  uint32_t repl;
  struct GNUNET_SQ_QueryParam params_sel_repl[] =
  { GNUNET_SQ_query_param_uint64(&rvalue),
    GNUNET_SQ_query_param_uint32(&repl),
    GNUNET_SQ_query_param_end };
  struct GNUNET_SQ_QueryParam params_upd_repl[] =
  { GNUNET_SQ_query_param_uint64(&rc.uid), GNUNET_SQ_query_param_end };

  GNUNET_log_from(GNUNET_ERROR_TYPE_DEBUG,
                  "datastore-sqlite",
                  "Getting random block based on replication order.\n");
  if (SQLITE_ROW != sqlite3_step(plugin->maxRepl))
    {
      GNUNET_SQ_reset(plugin->dbh, plugin->maxRepl);
      /* DB empty */
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  repl = sqlite3_column_int(plugin->maxRepl, 0);
  GNUNET_SQ_reset(plugin->dbh, plugin->maxRepl);
  rvalue = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
  if (GNUNET_OK != GNUNET_SQ_bind(plugin->selRepl, params_sel_repl))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  rc.have_uid = GNUNET_SYSERR;
  rc.proc = proc;
  rc.proc_cls = proc_cls;
  execute_get(plugin, plugin->selRepl, &repl_proc, &rc);
  if (GNUNET_YES == rc.have_uid)
    {
      if (GNUNET_OK != GNUNET_SQ_bind(plugin->updRepl, params_upd_repl))
        {
          proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
          return;
        }
      if (SQLITE_DONE != sqlite3_step(plugin->updRepl))
        LOG_SQLITE(plugin,
                   GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                   "sqlite3_step");
      GNUNET_SQ_reset(plugin->dbh, plugin->updRepl);
    }
  if (GNUNET_SYSERR == rc.have_uid)
    {
      /* proc was not called at all so far, do it now. */
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
    }
}


static void
sqlite_plugin_get_expiration(void *cls,
                             PluginDatumProcessor proc,
                             void *proc_cls)
{
  struct Plugin *plugin = cls;
  sqlite3_stmt *stmt;
  struct GNUNET_TIME_Absolute now;
  struct GNUNET_SQ_QueryParam params[] = { GNUNET_SQ_query_param_absolute_time(
                                             &now),
                                           GNUNET_SQ_query_param_end };

  GNUNET_log_from(
    GNUNET_ERROR_TYPE_DEBUG,
    "sqlite",
    "Getting random block based on expiration and priority order.\n");
  now = GNUNET_TIME_absolute_get();
  stmt = plugin->selExpi;
  if (GNUNET_OK != GNUNET_SQ_bind(stmt, params))
    {
      proc(proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
      return;
    }
  execute_get(plugin, stmt, proc, proc_cls);
}


static void
sqlite_plugin_get_keys(void *cls, PluginKeyProcessor proc, void *proc_cls)
{
  struct Plugin *plugin = cls;
  struct GNUNET_HashCode key;
  struct GNUNET_SQ_ResultSpec results[] =
  { GNUNET_SQ_result_spec_auto_from_type(&key), GNUNET_SQ_result_spec_end };
  sqlite3_stmt *stmt;
  int ret;

  GNUNET_assert(NULL != proc);
  if (SQLITE_OK != sq_prepare(plugin->dbh, "SELECT hash FROM gn091", &stmt))
    {
      LOG_SQLITE(plugin,
                 GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite_prepare");
      proc(proc_cls, NULL, 0);
      return;
    }
  while (SQLITE_ROW == (ret = sqlite3_step(stmt)))
    {
      if (GNUNET_OK == GNUNET_SQ_extract_result(stmt, results))
        proc(proc_cls, &key, 1);
      else
        GNUNET_break(0);
    }
  if (SQLITE_DONE != ret)
    LOG_SQLITE(plugin, GNUNET_ERROR_TYPE_ERROR, "sqlite_step");
  sqlite3_finalize(stmt);
  proc(proc_cls, NULL, 0);
}


static void
sqlite_plugin_drop(void *cls)
{
  struct Plugin *plugin = cls;

  plugin->drop_on_shutdown = GNUNET_YES;
}


static void
sqlite_plugin_remove_key(void *cls,
                         const struct GNUNET_HashCode *key,
                         uint32_t size,
                         const void *data,
                         PluginRemoveCont cont,
                         void *cont_cls)
{
  struct Plugin *plugin = cls;
  struct GNUNET_SQ_QueryParam params[] =
  { GNUNET_SQ_query_param_auto_from_type(key),
    GNUNET_SQ_query_param_fixed_size(data, size),
    GNUNET_SQ_query_param_end };

  if (GNUNET_OK != GNUNET_SQ_bind(plugin->remove, params))
    {
      cont(cont_cls, key, size, GNUNET_SYSERR, "bind failed");
      return;
    }
  if (SQLITE_DONE != sqlite3_step(plugin->remove))
    {
      LOG_SQLITE(plugin,
                 GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_step");
      GNUNET_SQ_reset(plugin->dbh, plugin->remove);
      cont(cont_cls, key, size, GNUNET_SYSERR, "sqlite3_step failed");
      return;
    }
  int changes = sqlite3_changes(plugin->dbh);
  GNUNET_SQ_reset(plugin->dbh, plugin->remove);
  if (0 == changes)
    {
      cont(cont_cls, key, size, GNUNET_NO, NULL);
      return;
    }
  if (NULL != plugin->env->duc)
    plugin->env->duc(plugin->env->cls,
                     -(size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
  cont(cont_cls, key, size, GNUNET_OK, NULL);
}


static void
sqlite_plugin_estimate_size(void *cls, unsigned long long *estimate)
{
  struct Plugin *plugin = cls;
  sqlite3_stmt *stmt;
  uint64_t pages;
  uint64_t page_size;

#if ENULL_DEFINED
  char *e;
#endif

  if (NULL == estimate)
    return;
  if (SQLITE_VERSION_NUMBER < 3006000)
    {
      GNUNET_log_from(
        GNUNET_ERROR_TYPE_WARNING,
        "datastore-sqlite",
        _("sqlite version to old to determine size, assuming zero\n"));
      *estimate = 0;
      return;
    }
  CHECK(SQLITE_OK == sqlite3_exec(plugin->dbh, "VACUUM", NULL, NULL, ENULL));
  CHECK(SQLITE_OK == sqlite3_exec(plugin->dbh,
                                  "PRAGMA auto_vacuum=INCREMENTAL",
                                  NULL,
                                  NULL,
                                  ENULL));
  CHECK(SQLITE_OK == sq_prepare(plugin->dbh, "PRAGMA page_count", &stmt));
  if (SQLITE_ROW == sqlite3_step(stmt))
    pages = sqlite3_column_int64(stmt, 0);
  else
    pages = 0;
  sqlite3_finalize(stmt);
  CHECK(SQLITE_OK == sq_prepare(plugin->dbh, "PRAGMA page_size", &stmt));
  CHECK(SQLITE_ROW == sqlite3_step(stmt));
  page_size = sqlite3_column_int64(stmt, 0);
  sqlite3_finalize(stmt);
  GNUNET_log(
    GNUNET_ERROR_TYPE_INFO,
    _(
      "Using sqlite page utilization to estimate payload (%llu pages of size %llu bytes)\n"),
    (unsigned long long)pages,
    (unsigned long long)page_size);
  *estimate = pages * page_size;
}


void *
libgnunet_plugin_datastore_sqlite_init(void *cls)
{
  static struct Plugin plugin;
  struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
  struct GNUNET_DATASTORE_PluginFunctions *api;

  if (NULL != plugin.env)
    return NULL; /* can only initialize once! */
  memset(&plugin, 0, sizeof(struct Plugin));
  plugin.env = env;
  if (GNUNET_OK != database_setup(env->cfg, &plugin))
    {
      database_shutdown(&plugin);
      return NULL;
    }
  api = GNUNET_new(struct GNUNET_DATASTORE_PluginFunctions);
  api->cls = &plugin;
  api->estimate_size = &sqlite_plugin_estimate_size;
  api->put = &sqlite_plugin_put;
  api->get_key = &sqlite_plugin_get_key;
  api->get_replication = &sqlite_plugin_get_replication;
  api->get_expiration = &sqlite_plugin_get_expiration;
  api->get_zero_anonymity = &sqlite_plugin_get_zero_anonymity;
  api->get_keys = &sqlite_plugin_get_keys;
  api->drop = &sqlite_plugin_drop;
  api->remove_key = &sqlite_plugin_remove_key;
  GNUNET_log_from(GNUNET_ERROR_TYPE_INFO,
                  "sqlite",
                  _("Sqlite database running\n"));
  return api;
}


void *
libgnunet_plugin_datastore_sqlite_done(void *cls)
{
  char *fn;
  struct GNUNET_DATASTORE_PluginFunctions *api = cls;
  struct Plugin *plugin = api->cls;

  GNUNET_log_from(GNUNET_ERROR_TYPE_DEBUG,
                  "sqlite",
                  "sqlite plugin is done\n");
  fn = NULL;
  if (plugin->drop_on_shutdown)
    fn = GNUNET_strdup(plugin->fn);
  database_shutdown(plugin);
  plugin->env = NULL;
  GNUNET_free(api);
  if (NULL != fn)
    {
      if (0 != unlink(fn))
        GNUNET_log_strerror_file(GNUNET_ERROR_TYPE_WARNING, "unlink", fn);
      GNUNET_free(fn);
    }
  return NULL;
}

/* end of plugin_datastore_sqlite.c */