gnunet-service-namestore.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2012, 2013, 2014, 2018 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_error_codes.h"
#include "gnunet_gnsrecord_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_util_lib.h"
#include "gnunet_namestore_plugin.h"
#include "gnunet_statistics_service.h"
#include "namestore.h"
 
#define LOG_STRERROR_FILE(kind, syscall, filename) \
        GNUNET_log_from_strerror_file (kind, "util", syscall, filename)
 
#define MONITOR_STALL_WARN_DELAY GNUNET_TIME_UNIT_MINUTES
 
#define NC_SIZE 16
 
struct NamestoreClient;
 
 
struct ZoneIteration
{
  struct ZoneIteration *next;
 
  struct ZoneIteration *prev;
 
  struct NamestoreClient *nc;
 
  struct GNUNET_GNSRECORD_Data *nick;
 
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
 
  enum GNUNET_GNSRECORD_Filter filter;
 
  uint64_t seq;
 
  uint32_t request_id;
 
  uint32_t offset;
 
  int send_end;
};
 
struct NamestoreClient
{
  struct GNUNET_SERVICE_Client *client;
 
  struct GNUNET_NAMESTORE_PluginFunctions *GSN_database;
 
  char *db_lib_name;
 
  struct GNUNET_MQ_Handle *mq;
 
  struct ZoneIteration *op_head;
 
  struct ZoneIteration *op_tail;
};
 
 
struct ZoneMonitor
{
  struct ZoneMonitor *next;
 
  struct ZoneMonitor *prev;
 
  struct NamestoreClient *nc;
 
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
 
  enum GNUNET_GNSRECORD_Filter filter;
 
  struct GNUNET_SCHEDULER_Task *task;
 
  struct GNUNET_SCHEDULER_Task *sa_wait_warning;
 
  struct GNUNET_TIME_Absolute sa_waiting_start;
 
  uint64_t seq;
 
  uint64_t limit;
 
  uint64_t iteration_cnt;
 
  int in_first_iteration;
 
  int run_again;
 
  int sa_waiting;
};
 
 
struct StoreActivity
{
  struct StoreActivity *next;
 
  struct StoreActivity *prev;
 
  struct NamestoreClient *nc;
 
  uint32_t rid;
 
  uint16_t rd_set_pos;
 
  uint16_t rd_set_count;
 
  struct GNUNET_CRYPTO_BlindablePrivateKey private_key;
 
  const struct RecordSet *rs;
 
  struct ZoneMonitor *zm_pos;
 
};
 
 
struct NickCache
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
 
  struct GNUNET_GNSRECORD_Data *rd;
 
  struct GNUNET_TIME_Absolute last_used;
};
 
static struct NickCache nick_cache[NC_SIZE];
 
static const struct GNUNET_CRYPTO_BlindablePrivateKey zero;
 
static const struct GNUNET_CONFIGURATION_Handle *GSN_cfg;
 
static struct GNUNET_STATISTICS_Handle *statistics;
 
static char *db_lib_name;
 
struct GNUNET_NAMESTORE_PluginFunctions *GSN_database;
 
 
static struct ZoneMonitor *monitor_head;
 
static struct ZoneMonitor *monitor_tail;
 
static struct StoreActivity *sa_head;
 
static struct StoreActivity *sa_tail;
 
static struct GNUNET_NotificationContext *monitor_nc;
 
static int return_orphaned;
 
static void
cleanup_task (void *cls)
{
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stopping namestore service\n");
  if (NULL != monitor_nc)
  {
    GNUNET_notification_context_destroy (monitor_nc);
    monitor_nc = NULL;
  }
  if (NULL != statistics)
  {
    GNUNET_STATISTICS_destroy (statistics, GNUNET_NO);
    statistics = NULL;
  }
  GNUNET_break (NULL == GNUNET_PLUGIN_unload (db_lib_name, GSN_database));
  GNUNET_free (db_lib_name);
  db_lib_name = NULL;
}
 
 
static void
free_store_activity (struct StoreActivity *sa)
{
  GNUNET_CONTAINER_DLL_remove (sa_head, sa_tail, sa);
  GNUNET_free (sa);
}
 
 
static void
lookup_nick_it (void *cls,
                uint64_t seq,
                const char *editor_hint,
                const struct GNUNET_CRYPTO_BlindablePrivateKey *private_key,
                const char *label,
                unsigned int rd_count,
                const struct GNUNET_GNSRECORD_Data *rd)
{
  struct GNUNET_GNSRECORD_Data **res = cls;
 
  (void) private_key;
  GNUNET_assert (0 != seq);
  if (0 != strcmp (label, GNUNET_GNS_EMPTY_LABEL_AT))
  {
    GNUNET_break (0);
    return;
  }
  for (unsigned int c = 0; c < rd_count; c++)
  {
    if (GNUNET_GNSRECORD_TYPE_NICK == rd[c].record_type)
    {
      (*res) =
        GNUNET_malloc (rd[c].data_size + sizeof(struct GNUNET_GNSRECORD_Data));
      (*res)->data = &(*res)[1];
      GNUNET_memcpy ((void *) (*res)->data, rd[c].data, rd[c].data_size);
      (*res)->data_size = rd[c].data_size;
      (*res)->expiration_time = rd[c].expiration_time;
      (*res)->flags = rd[c].flags;
      (*res)->record_type = GNUNET_GNSRECORD_TYPE_NICK;
      return;
    }
  }
  (*res) = NULL;
}
 
 
static void
cache_nick (const struct GNUNET_CRYPTO_BlindablePrivateKey *zone,
            const struct GNUNET_GNSRECORD_Data *nick)
{
  struct NickCache *oldest;
 
  oldest = NULL;
  for (unsigned int i = 0; i < NC_SIZE; i++)
  {
    struct NickCache *pos = &nick_cache[i];
 
    if ((NULL == oldest) ||
        (oldest->last_used.abs_value_us > pos->last_used.abs_value_us))
      oldest = pos;
    if (0 == GNUNET_memcmp (zone, &pos->zone))
    {
      oldest = pos;
      break;
    }
  }
  GNUNET_free (oldest->rd);
  oldest->zone = *zone;
  if (NULL != nick)
  {
    oldest->rd = GNUNET_malloc (sizeof(*nick) + nick->data_size);
    *oldest->rd = *nick;
    oldest->rd->data = &oldest->rd[1];
    memcpy (&oldest->rd[1], nick->data, nick->data_size);
  }
  else
  {
    oldest->rd = NULL;
  }
  oldest->last_used = GNUNET_TIME_absolute_get ();
}
 
 
static struct GNUNET_GNSRECORD_Data *
get_nick_record (const struct GNUNET_CRYPTO_BlindablePrivateKey *zone)
{
  struct GNUNET_CRYPTO_BlindablePublicKey pub;
  struct GNUNET_GNSRECORD_Data *nick;
  int res;
 
  /* check cache first */
  for (unsigned int i = 0; i < NC_SIZE; i++)
  {
    struct NickCache *pos = &nick_cache[i];
    if ((NULL != pos->rd) && (0 == GNUNET_memcmp (zone, &pos->zone)))
    {
      if (NULL == pos->rd)
        return NULL;
      nick = GNUNET_malloc (sizeof(*nick) + pos->rd->data_size);
      *nick = *pos->rd;
      nick->data = &nick[1];
      memcpy (&nick[1], pos->rd->data, pos->rd->data_size);
      pos->last_used = GNUNET_TIME_absolute_get ();
      return nick;
    }
  }
 
  nick = NULL;
  res = GSN_database->lookup_records (GSN_database->cls,
                                      zone,
                                      GNUNET_GNS_EMPTY_LABEL_AT,
                                      &lookup_nick_it,
                                      &nick);
  if ((GNUNET_OK != res) || (NULL == nick))
  {
#if ! defined(GNUNET_CULL_LOGGING)
    static int do_log = GNUNET_LOG_CALL_STATUS;
 
    if (0 == do_log)
      do_log = GNUNET_get_log_call_status (GNUNET_ERROR_TYPE_DEBUG,
                                           "namestore",
                                           __FILE__,
                                           __FUNCTION__,
                                           __LINE__);
    if (1 == do_log)
    {
      GNUNET_CRYPTO_blindable_key_get_public (zone, &pub);
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
                  "No nick name set for zone `%s'\n",
                  GNUNET_GNSRECORD_z2s (&pub));
    }
#endif
    /* update cache */
    cache_nick (zone, NULL);
    return NULL;
  }
 
  /* update cache */
  cache_nick (zone, nick);
  return nick;
}
 
 
static void
merge_with_nick_records (const struct GNUNET_GNSRECORD_Data *nick_rd,
                         unsigned int rd2_length,
                         const struct GNUNET_GNSRECORD_Data *rd2,
                         unsigned int *rdc_res,
                         struct GNUNET_GNSRECORD_Data **rd_res)
{
  uint64_t latest_expiration;
  size_t req;
  char *data;
  size_t data_offset;
  struct GNUNET_GNSRECORD_Data *target;
 
  (*rdc_res) = 1 + rd2_length;
  if (0 == 1 + rd2_length)
  {
    GNUNET_break (0);
    (*rd_res) = NULL;
    return;
  }
  req = sizeof(struct GNUNET_GNSRECORD_Data) + nick_rd->data_size;
  for (unsigned int i = 0; i < rd2_length; i++)
  {
    const struct GNUNET_GNSRECORD_Data *orig = &rd2[i];
 
    if (req + sizeof(struct GNUNET_GNSRECORD_Data) + orig->data_size < req)
    {
      GNUNET_break (0);
      (*rd_res) = NULL;
      return;
    }
    req += sizeof(struct GNUNET_GNSRECORD_Data) + orig->data_size;
  }
  target = GNUNET_malloc (req);
  (*rd_res) = target;
  data = (char *) &target[1 + rd2_length];
  data_offset = 0;
  latest_expiration = 0;
  for (unsigned int i = 0; i < rd2_length; i++)
  {
    const struct GNUNET_GNSRECORD_Data *orig = &rd2[i];
 
    if (0 != (orig->flags & GNUNET_GNSRECORD_RF_RELATIVE_EXPIRATION))
    {
      if ((GNUNET_TIME_absolute_get ().abs_value_us + orig->expiration_time) >
          latest_expiration)
        latest_expiration = orig->expiration_time;
    }
    else if (orig->expiration_time > latest_expiration)
      latest_expiration = orig->expiration_time;
    target[i] = *orig;
    target[i].data = (void *) &data[data_offset];
    GNUNET_memcpy (&data[data_offset], orig->data, orig->data_size);
    data_offset += orig->data_size;
  }
  /* append nick */
  target[rd2_length] = *nick_rd;
  /* Mark as supplemental */
  target[rd2_length].flags = nick_rd->flags | GNUNET_GNSRECORD_RF_SUPPLEMENTAL;
  target[rd2_length].expiration_time = latest_expiration;
  target[rd2_length].data = (void *) &data[data_offset];
  GNUNET_memcpy (&data[data_offset], nick_rd->data, nick_rd->data_size);
  data_offset += nick_rd->data_size;
  GNUNET_assert (req == (sizeof(struct GNUNET_GNSRECORD_Data)) * (*rdc_res)
                 + data_offset);
}
 
 
static int
send_lookup_response_with_filter (struct NamestoreClient *nc,
                                  uint32_t request_id,
                                  const struct
                                  GNUNET_CRYPTO_BlindablePrivateKey *zone_key,
                                  const char *name,
                                  unsigned int rd_count,
                                  const struct GNUNET_GNSRECORD_Data *rd,
                                  enum GNUNET_GNSRECORD_Filter filter)
{
  struct GNUNET_MQ_Envelope *env;
  struct RecordResultMessage *zir_msg;
  struct GNUNET_GNSRECORD_Data *nick;
  struct GNUNET_GNSRECORD_Data *res;
  struct GNUNET_GNSRECORD_Data rd_nf[rd_count];
  struct GNUNET_TIME_Absolute block_exp;
  unsigned int res_count;
  unsigned int rd_nf_count;
  size_t name_len;
  size_t key_len;
  ssize_t rd_ser_len;
  char *name_tmp;
  char *rd_ser;
  char *emsg;
 
  block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
  nick = get_nick_record (zone_key);
  GNUNET_assert (-1 != GNUNET_GNSRECORD_records_get_size (rd_count, rd));
 
  if (GNUNET_OK != GNUNET_GNSRECORD_normalize_record_set (name,
                                                          rd,
                                                          rd_count,
                                                          rd_nf,
                                                          &rd_nf_count,
                                                          &block_exp,
                                                          filter,
                                                          &emsg))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "%s\n", emsg);
    GNUNET_free (emsg);
    GNUNET_assert (0);
  }
 
  if ((NULL != nick) && (0 != strcmp (name, GNUNET_GNS_EMPTY_LABEL_AT)))
  {
    nick->flags =
      (nick->flags | GNUNET_GNSRECORD_RF_PRIVATE) ^ GNUNET_GNSRECORD_RF_PRIVATE;
    merge_with_nick_records (nick, rd_nf_count, rd_nf, &res_count, &res);
  }
  else
  {
    res_count = rd_nf_count;
    res = (struct GNUNET_GNSRECORD_Data *) rd_nf;
  }
  if (NULL != nick)
    GNUNET_free (nick);
 
  if (0 == res_count)
  {
    if (rd_nf != res)
      GNUNET_free (res);
    return 0;
  }
  GNUNET_assert (-1 != GNUNET_GNSRECORD_records_get_size (res_count, res));
 
 
  name_len = strlen (name) + 1;
  rd_ser_len = GNUNET_GNSRECORD_records_get_size (res_count, res);
  if (rd_ser_len < 0)
  {
    if (rd_nf != res)
      GNUNET_free (res);
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return 0;
  }
  if (((size_t) rd_ser_len) >= UINT16_MAX - name_len - sizeof(*zir_msg))
  {
    if (rd_nf != res)
      GNUNET_free (res);
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return 0;
  }
  key_len = GNUNET_CRYPTO_blindable_sk_get_length (zone_key);
  env = GNUNET_MQ_msg_extra (zir_msg,
                             name_len + rd_ser_len + key_len,
                             GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_RESULT);
  zir_msg->gns_header.r_id = htonl (request_id);
  zir_msg->name_len = htons (name_len);
  zir_msg->rd_count = htons (res_count);
  zir_msg->rd_len = htons ((uint16_t) rd_ser_len);
  zir_msg->key_len = htons (key_len);
  GNUNET_CRYPTO_write_blindable_sk_to_buffer (zone_key,
                                              &zir_msg[1],
                                              key_len);
  zir_msg->expire = GNUNET_TIME_absolute_hton (block_exp);
  name_tmp = (char *) &zir_msg[1] + key_len;
  GNUNET_memcpy (name_tmp, name, name_len);
  rd_ser = &name_tmp[name_len];
  GNUNET_assert (
    rd_ser_len ==
    GNUNET_GNSRECORD_records_serialize (res_count, res, rd_ser_len, rd_ser));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending RECORD_RESULT message with %u records\n",
              res_count);
  GNUNET_STATISTICS_update (statistics,
                            "Record sets sent to clients",
                            1,
                            GNUNET_NO);
  GNUNET_MQ_send (nc->mq, env);
  if (rd_nf != res)
    GNUNET_free (res);
  return res_count;
}
 
 
static void
send_store_response (struct NamestoreClient *nc,
                     enum GNUNET_ErrorCode ec,
                     uint32_t rid)
{
  struct GNUNET_MQ_Envelope *env;
  struct NamestoreResponseMessage *rcr_msg;
 
  GNUNET_assert (NULL != nc);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending GENERIC_RESPONSE message\n");
  GNUNET_STATISTICS_update (statistics,
                            "Store requests completed",
                            1,
                            GNUNET_NO);
  env = GNUNET_MQ_msg (rcr_msg,
                       GNUNET_MESSAGE_TYPE_NAMESTORE_GENERIC_RESPONSE);
  rcr_msg->gns_header.r_id = htonl (rid);
  rcr_msg->ec = htonl (ec);
  GNUNET_MQ_send (nc->mq, env);
}
 
 
static void
zone_iteration_done_client_continue (struct ZoneIteration *zi)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_NAMESTORE_Header *em;
 
  GNUNET_SERVICE_client_continue (zi->nc->client);
  if (! zi->send_end)
    return;
  /* send empty response to indicate end of list */
  env = GNUNET_MQ_msg (em, GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_RESULT_END);
  em->r_id = htonl (zi->request_id);
  GNUNET_MQ_send (zi->nc->mq, env);
 
  GNUNET_CONTAINER_DLL_remove (zi->nc->op_head, zi->nc->op_tail, zi);
  GNUNET_free (zi);
}
 
 
static void
warn_monitor_slow (void *cls)
{
  struct ZoneMonitor *zm = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
              "No response from monitor since %s\n",
              GNUNET_STRINGS_absolute_time_to_string (zm->sa_waiting_start));
  zm->sa_wait_warning = GNUNET_SCHEDULER_add_delayed (MONITOR_STALL_WARN_DELAY,
                                                      &warn_monitor_slow,
                                                      zm);
}
 
 
static int
continue_store_activity (struct StoreActivity *sa,
                         int call_continue)
{
  const struct RecordSet *rd_set = sa->rs;
  unsigned int rd_count;
  size_t name_len;
  size_t rd_ser_len;
  const char *name;
  const char *rd_ser;
  const char *buf;
 
  buf = (const char *) &sa[1];
  for (int i = sa->rd_set_pos; i < sa->rd_set_count; i++)
  {
    rd_set = (struct RecordSet *) buf;
    name_len = ntohs (rd_set->name_len);
    rd_count = ntohs (rd_set->rd_count);
    rd_ser_len = ntohs (rd_set->rd_len);
    name = (const char *) &rd_set[1];
    buf += sizeof (struct RecordSet) + name_len + rd_ser_len;
    rd_ser = &name[name_len];
    {
      struct GNUNET_GNSRECORD_Data rd[GNUNET_NZL (rd_count)];
 
      /* We did this before, must succeed again */
      GNUNET_assert (
        GNUNET_OK ==
        GNUNET_GNSRECORD_records_deserialize (rd_ser_len, rd_ser, rd_count,
                                              rd));
 
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Checking monitors watching for `%s'\n",
                  name);
      for (struct ZoneMonitor *zm = sa->zm_pos; NULL != zm; zm = sa->zm_pos)
      {
        if ((0 != GNUNET_memcmp (&sa->private_key, &zm->zone)) &&
            (0 != GNUNET_memcmp (&zm->zone, &zero)))
        {
          sa->zm_pos = zm->next;   /* not interesting to this monitor */
          continue;
        }
        if (zm->limit == zm->iteration_cnt)
        {
          zm->sa_waiting = GNUNET_YES;
          zm->sa_waiting_start = GNUNET_TIME_absolute_get ();
          if (NULL != zm->sa_wait_warning)
            GNUNET_SCHEDULER_cancel (zm->sa_wait_warning);
          zm->sa_wait_warning =
            GNUNET_SCHEDULER_add_delayed (MONITOR_STALL_WARN_DELAY,
                                          &warn_monitor_slow,
                                          zm);
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                      "Monitor is blocking client for `%s'\n",
                      name);
          return GNUNET_NO;    /* blocked on zone monitor */
        }
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Notifying monitor about changes under label `%s'\n",
                    name);
        if (0 < send_lookup_response_with_filter (zm->nc,
                                                  0,
                                                  &sa->private_key,
                                                  name,
                                                  rd_count,
                                                  rd,
                                                  zm->filter))
          zm->limit--;
        sa->zm_pos = zm->next;
      }
      sa->rd_set_pos++;
    }
  }
  if (GNUNET_YES == call_continue)
    GNUNET_SERVICE_client_continue (sa->nc->client);
  send_store_response (sa->nc, GNUNET_EC_NONE, sa->rid);
  free_store_activity (sa);
  return GNUNET_OK;
}
 
 
static void
client_disconnect_cb (void *cls,
                      struct GNUNET_SERVICE_Client *client,
                      void *app_ctx)
{
  struct NamestoreClient *nc = app_ctx;
  struct ZoneIteration *no;
  struct StoreActivity *sa = sa_head;
  struct StoreActivity *sn;
 
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p disconnected\n", client);
  for (struct ZoneMonitor *zm = monitor_head; NULL != zm; zm = zm->next)
  {
    if (nc != zm->nc)
      continue;
    GNUNET_CONTAINER_DLL_remove (monitor_head, monitor_tail, zm);
    if (NULL != zm->task)
    {
      GNUNET_SCHEDULER_cancel (zm->task);
      zm->task = NULL;
    }
    if (NULL != zm->sa_wait_warning)
    {
      GNUNET_SCHEDULER_cancel (zm->sa_wait_warning);
      zm->sa_wait_warning = NULL;
    }
    for (sa = sa_head; NULL != sa; sa = sn)
    {
      sn = sa->next;
      if (zm == sa->zm_pos)
      {
        sa->zm_pos = zm->next;
        /* this may free sa */
        continue_store_activity (sa, GNUNET_YES);
      }
    }
    GNUNET_free (zm);
    break;
  }
  sa = sa_head;
  while (NULL != sa)
  {
    if (nc != sa->nc)
    {
      sa = sa->next;
      continue;
    }
    sn = sa->next;
    free_store_activity (sa);
    sa = sn;
  }
  while (NULL != (no = nc->op_head))
  {
    GNUNET_CONTAINER_DLL_remove (nc->op_head, nc->op_tail, no);
    GNUNET_free (no);
  }
  GNUNET_break (NULL == GNUNET_PLUGIN_unload (nc->db_lib_name,
                                              nc->GSN_database));
  GNUNET_free (nc->db_lib_name);
  GNUNET_free (nc);
}
 
 
static void *
client_connect_cb (void *cls,
                   struct GNUNET_SERVICE_Client *client,
                   struct GNUNET_MQ_Handle *mq)
{
  struct NamestoreClient *nc;
  char *database;
 
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p connected\n", client);
  nc = GNUNET_new (struct NamestoreClient);
  nc->client = client;
  nc->mq = mq;
  /* Loading database plugin */
  if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_string (GSN_cfg,
                                                          "namestore",
                                                          "database",
                                                          &database))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "No database backend configured\n");
    GNUNET_free (nc);
    return NULL;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Loading %s\n",
              db_lib_name);
  nc->GSN_database = GNUNET_PLUGIN_load (GNUNET_OS_project_data_gnunet (),
                                         db_lib_name,
                                         (void *) GSN_cfg);
  GNUNET_free (database);
  if (NULL == nc->GSN_database)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Could not load database backend `%s'\n",
                db_lib_name);
    GNUNET_free (nc);
    return NULL;
  }
  nc->db_lib_name = GNUNET_strdup (db_lib_name);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Loaded %s\n",
              db_lib_name);
  return nc;
}
 
 
struct RecordLookupContext
{
  const char *label;
 
  char *editor_hint;
 
  char *res_rd;
 
  struct GNUNET_GNSRECORD_Data *nick;
 
  int found;
 
  enum GNUNET_GNSRECORD_Filter filter;
 
  unsigned int res_rd_count;
 
  ssize_t rd_ser_len;
};
 
 
static void
lookup_it (void *cls,
           uint64_t seq,
           const char *editor_hint,
           const struct GNUNET_CRYPTO_BlindablePrivateKey *private_key,
           const char *label,
           unsigned int rd_count_nf,
           const struct GNUNET_GNSRECORD_Data *rd_nf)
{
  struct RecordLookupContext *rlc = cls;
  struct GNUNET_GNSRECORD_Data rd[rd_count_nf];
  struct GNUNET_TIME_Absolute block_exp;
  unsigned int rd_count = 0;
  char *emsg;
 
  (void) private_key;
  GNUNET_assert (0 != seq);
  if (0 != strcmp (label, rlc->label))
    return;
  rlc->found = GNUNET_YES;
  if (NULL == rlc->editor_hint)
    rlc->editor_hint = GNUNET_strdup (editor_hint);
  if (GNUNET_OK != GNUNET_GNSRECORD_normalize_record_set (rlc->label,
                                                          rd_nf,
                                                          rd_count_nf,
                                                          rd,
                                                          &rd_count,
                                                          &block_exp,
                                                          rlc->filter,
                                                          &emsg))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "%s\n", emsg);
    GNUNET_free (emsg);
    GNUNET_assert (0);
  }
 
  if (0 == rd_count)
  {
    rlc->rd_ser_len = 0;
    rlc->res_rd_count = 0;
    rlc->res_rd = NULL;
    return;
  }
  if ((NULL != rlc->nick) && (0 != strcmp (label, GNUNET_GNS_EMPTY_LABEL_AT)))
  {
    /* Merge */
    struct GNUNET_GNSRECORD_Data *rd_res;
    unsigned int rdc_res;
 
    rd_res = NULL;
    rdc_res = 0;
    rlc->nick->flags = (rlc->nick->flags | GNUNET_GNSRECORD_RF_PRIVATE)
                       ^ GNUNET_GNSRECORD_RF_PRIVATE;
    merge_with_nick_records (rlc->nick, rd_count, rd, &rdc_res, &rd_res);
    rlc->rd_ser_len = GNUNET_GNSRECORD_records_get_size (rdc_res, rd_res);
    if (rlc->rd_ser_len < 0)
    {
      GNUNET_break (0);
      GNUNET_free (rd_res);
      rlc->found = GNUNET_NO;
      rlc->rd_ser_len = 0;
      return;
    }
    rlc->res_rd_count = rdc_res;
    rlc->res_rd = GNUNET_malloc (rlc->rd_ser_len);
    if (rlc->rd_ser_len != GNUNET_GNSRECORD_records_serialize (rdc_res,
                                                               rd_res,
                                                               rlc->rd_ser_len,
                                                               rlc->res_rd))
    {
      GNUNET_break (0);
      GNUNET_free (rlc->res_rd);
      rlc->res_rd = NULL;
      rlc->res_rd_count = 0;
      rlc->rd_ser_len = 0;
      GNUNET_free (rd_res);
      rlc->found = GNUNET_NO;
      return;
    }
    GNUNET_free (rd_res);
    GNUNET_free (rlc->nick);
    rlc->nick = NULL;
  }
  else
  {
    rlc->rd_ser_len = GNUNET_GNSRECORD_records_get_size (rd_count, rd);
    if (rlc->rd_ser_len < 0)
    {
      GNUNET_break (0);
      rlc->found = GNUNET_NO;
      rlc->rd_ser_len = 0;
      return;
    }
    rlc->res_rd_count = rd_count;
    rlc->res_rd = GNUNET_malloc (rlc->rd_ser_len);
    if (rlc->rd_ser_len != GNUNET_GNSRECORD_records_serialize (rd_count,
                                                               rd,
                                                               rlc->rd_ser_len,
                                                               rlc->res_rd))
    {
      GNUNET_break (0);
      GNUNET_free (rlc->res_rd);
      rlc->res_rd = NULL;
      rlc->res_rd_count = 0;
      rlc->rd_ser_len = 0;
      rlc->found = GNUNET_NO;
      return;
    }
  }
}
 
 
static int
check_edit_record_set (void *cls, const struct EditRecordSetMessage *er_msg)
{
  uint16_t name_len;
  uint16_t editor_hint_len;
  size_t src_size;
  size_t key_len;
 
  (void) cls;
  name_len = ntohs (er_msg->label_len);
  editor_hint_len = ntohs (er_msg->editor_hint_len);
  key_len = ntohs (er_msg->key_len);
  src_size = ntohs (er_msg->gns_header.header.size);
  if (name_len + editor_hint_len + key_len != src_size - sizeof(struct
                                                                EditRecordSetMessage))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_edit_record_set (void *cls, const struct EditRecordSetMessage *er_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  struct GNUNET_MQ_Envelope *env;
  struct EditRecordSetResponseMessage *rer_msg;
  struct RecordLookupContext rlc;
  const char *name_tmp;
  const char *editor_hint;
  char *conv_name;
  uint16_t name_len;
  uint16_t old_editor_hint_len;
  int res;
  size_t key_len;
  size_t kb_read;
 
  key_len = ntohs (er_msg->key_len);
  name_len = ntohs (er_msg->label_len);
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_private_key_from_buffer (&er_msg[1],
                                                   key_len,
                                                   &zone,
                                                   &kb_read)) ||
      (kb_read != key_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error reading private key\n");
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_tmp = (const char *) &er_msg[1] + key_len;
  editor_hint = (const char *) name_tmp + name_len;
  GNUNET_SERVICE_client_continue (nc->client);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received NAMESTORE_RECORD_SET_EDIT message for name `%s'\n",
              name_tmp);
 
  conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
  if (NULL == conv_name)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error converting name `%s'\n",
                name_tmp);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_len = strlen (conv_name) + 1;
  rlc.editor_hint = NULL;
  rlc.label = conv_name;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Looking up without filter\n");
  rlc.filter = GNUNET_GNSRECORD_FILTER_NONE;
  rlc.found = GNUNET_NO;
  rlc.res_rd_count = 0;
  rlc.res_rd = NULL;
  rlc.rd_ser_len = 0;
  rlc.nick = get_nick_record (&zone);
  res = nc->GSN_database->edit_records (nc->GSN_database->cls,
                                        editor_hint,
                                        &zone,
                                        conv_name,
                                        &lookup_it,
                                        &rlc);
 
  old_editor_hint_len = 0;
  if (NULL != rlc.editor_hint)
    old_editor_hint_len = strlen (rlc.editor_hint) + 1;
  env =
    GNUNET_MQ_msg_extra (rer_msg,
                         rlc.rd_ser_len + old_editor_hint_len,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_SET_EDIT_RESPONSE)
  ;
  rer_msg->editor_hint_len = htons (old_editor_hint_len);
  rer_msg->gns_header.r_id = er_msg->gns_header.r_id;
  rer_msg->rd_count = htons (rlc.res_rd_count);
  rer_msg->rd_len = htons (rlc.rd_ser_len);
  if (GNUNET_YES == rlc.found)
    rer_msg->ec = htons (GNUNET_EC_NONE);
  else if (GNUNET_SYSERR == res)
    rer_msg->ec = htons (GNUNET_EC_NAMESTORE_UNKNOWN);
  else
    rer_msg->ec = htons (GNUNET_EC_NAMESTORE_NO_RESULTS);
  GNUNET_memcpy (&rer_msg[1], rlc.editor_hint, old_editor_hint_len);
  GNUNET_memcpy ((char*) &rer_msg[1] + old_editor_hint_len, rlc.res_rd,
                 rlc.rd_ser_len);
  GNUNET_MQ_send (nc->mq, env);
  GNUNET_free (rlc.editor_hint);
  GNUNET_free (rlc.res_rd);
  GNUNET_free (conv_name);
}
 
 
static int
check_edit_record_set_cancel (void *cls, const struct
                              EditRecordSetCancelMessage *er_msg)
{
  uint16_t name_len;
  uint16_t editor_hint_len;
  uint16_t editor_hint_repl_len;
  size_t src_size;
  size_t key_len;
 
  (void) cls;
  name_len = ntohs (er_msg->label_len);
  editor_hint_len = ntohs (er_msg->editor_hint_len);
  editor_hint_repl_len = ntohs (er_msg->editor_hint_len);
  key_len = ntohs (er_msg->key_len);
  src_size = ntohs (er_msg->gns_header.header.size);
  if (name_len + editor_hint_len + editor_hint_repl_len + key_len != src_size
      - sizeof(struct
               EditRecordSetCancelMessage))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_edit_record_set_cancel (void *cls, const struct
                               EditRecordSetCancelMessage *er_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  struct GNUNET_MQ_Envelope *env;
  struct NamestoreResponseMessage *rer_msg;
  const char *name_tmp;
  const char *editor_hint;
  const char *editor_hint_repl;
  char *conv_name;
  uint16_t name_len;
  uint16_t editor_hint_len;
  int res;
  size_t key_len;
  size_t kb_read;
 
  key_len = ntohs (er_msg->key_len);
  name_len = ntohs (er_msg->label_len);
  editor_hint_len = ntohs (er_msg->editor_hint_len);
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_private_key_from_buffer (&er_msg[1],
                                                   key_len,
                                                   &zone,
                                                   &kb_read)) ||
      (kb_read != key_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error reading private key\n");
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_tmp = (const char *) &er_msg[1] + key_len;
  editor_hint = (const char *) name_tmp + name_len;
  editor_hint_repl = (const char *) name_tmp + name_len + editor_hint_len;
  GNUNET_SERVICE_client_continue (nc->client);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received NAMESTORE_RECORD_SET_EDIT message for name `%s'\n",
              name_tmp);
 
  conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
  if (NULL == conv_name)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error converting name `%s'\n",
                name_tmp);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_len = strlen (conv_name) + 1;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Clearing editor hint\n");
  res = nc->GSN_database->clear_editor_hint (nc->GSN_database->cls,
                                             editor_hint,
                                             editor_hint_repl,
                                             &zone,
                                             conv_name);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Clearing editor hint %s\n", (GNUNET_SYSERR == res) ? "failed." :
              "successful.");
 
  env =
    GNUNET_MQ_msg (rer_msg,
                   GNUNET_MESSAGE_TYPE_NAMESTORE_GENERIC_RESPONSE);
  rer_msg->gns_header.r_id = er_msg->gns_header.r_id;
  rer_msg->ec = htons ((GNUNET_OK == res) ? GNUNET_EC_NONE :
                       GNUNET_EC_NAMESTORE_BACKEND_FAILED);
  GNUNET_MQ_send (nc->mq, env);
  GNUNET_free (conv_name);
}
 
 
static int
check_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
{
  uint32_t name_len;
  size_t src_size;
  size_t key_len;
 
  (void) cls;
  name_len = ntohs (ll_msg->label_len);
  key_len = ntohs (ll_msg->key_len);
  src_size = ntohs (ll_msg->gns_header.header.size);
  if (name_len + key_len != src_size - sizeof(struct LabelLookupMessage))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  struct GNUNET_MQ_Envelope *env;
  struct LabelLookupResponseMessage *llr_msg;
  struct RecordLookupContext rlc;
  const char *name_tmp;
  char *res_name;
  char *conv_name;
  uint32_t name_len;
  int res;
  size_t key_len;
  size_t kb_read;
 
  key_len = ntohs (ll_msg->key_len);
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_private_key_from_buffer (&ll_msg[1],
                                                   key_len,
                                                   &zone,
                                                   &kb_read)) ||
      (kb_read != key_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error reading private key\n");
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_tmp = (const char *) &ll_msg[1] + key_len;
  GNUNET_SERVICE_client_continue (nc->client);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received NAMESTORE_RECORD_LOOKUP message for name `%s'\n",
              name_tmp);
 
  conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
  if (NULL == conv_name)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error converting name `%s'\n",
                name_tmp);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  name_len = strlen (conv_name) + 1;
  rlc.editor_hint = NULL;
  rlc.label = conv_name;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Looking up with filter %u\n", ntohs (ll_msg->filter));
  rlc.filter = ntohs (ll_msg->filter);
  rlc.found = GNUNET_NO;
  rlc.res_rd_count = 0;
  rlc.res_rd = NULL;
  rlc.rd_ser_len = 0;
  rlc.nick = get_nick_record (&zone);
  res = nc->GSN_database->lookup_records (nc->GSN_database->cls,
                                          &zone,
                                          conv_name,
                                          &lookup_it,
                                          &rlc);
  env =
    GNUNET_MQ_msg_extra (llr_msg,
                         key_len + name_len + rlc.rd_ser_len,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP_RESPONSE);
  llr_msg->gns_header.r_id = ll_msg->gns_header.r_id;
  GNUNET_memcpy (&llr_msg[1], &ll_msg[1], key_len);
  llr_msg->key_len = ll_msg->key_len;
  llr_msg->name_len = htons (name_len);
  llr_msg->rd_count = htons (rlc.res_rd_count);
  llr_msg->rd_len = htons (rlc.rd_ser_len);
  llr_msg->reserved = htons (0);
  res_name = ((char *) &llr_msg[1]) + key_len;
  if (GNUNET_YES == rlc.found)
    llr_msg->found = htons (GNUNET_YES);
  else if (GNUNET_SYSERR == res)
    llr_msg->found = htons (GNUNET_SYSERR);
  else
    llr_msg->found = htons (GNUNET_NO);
  GNUNET_memcpy (res_name, conv_name, name_len);
  GNUNET_memcpy (&res_name[name_len], rlc.res_rd, rlc.rd_ser_len);
  GNUNET_MQ_send (nc->mq, env);
  GNUNET_free (rlc.editor_hint);
  GNUNET_free (rlc.res_rd);
  GNUNET_free (conv_name);
}
 
 
static int
check_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
{
  size_t msg_size;
  size_t min_size_exp;
  size_t rd_set_count;
  size_t key_len;
 
  (void) cls;
  msg_size = ntohs (rp_msg->gns_header.header.size);
  rd_set_count = ntohs (rp_msg->rd_set_count);
  key_len = ntohs (rp_msg->key_len);
 
  min_size_exp = sizeof(*rp_msg) + key_len + sizeof (struct RecordSet)
                 * rd_set_count;
  if (msg_size < min_size_exp)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
struct LookupExistingRecordsContext
{
 
  struct GNUNET_TIME_Absolute exp;
 
  int only_tombstone;
 
};
 
 
static void
get_existing_rd_exp (void *cls,
                     uint64_t seq,
                     const char *editor_hint,
                     const struct
                     GNUNET_CRYPTO_BlindablePrivateKey *private_key,
                     const char *label,
                     unsigned int rd_count,
                     const struct GNUNET_GNSRECORD_Data *rd)
{
  struct LookupExistingRecordsContext *lctx = cls;
  struct GNUNET_GNSRECORD_Data rd_pub[rd_count];
  unsigned int rd_pub_count;
  char *emsg;
 
  if ((1 == rd_count) &&
      (GNUNET_GNSRECORD_TYPE_TOMBSTONE == rd[0].record_type))
  {
    /* This record set contains only a tombstone! */
    lctx->only_tombstone = GNUNET_YES;
  }
  if (GNUNET_OK !=
      GNUNET_GNSRECORD_normalize_record_set (label,
                                             rd,
                                             rd_count,
                                             rd_pub,
                                             &rd_pub_count,
                                             &lctx->exp,
                                             GNUNET_GNSRECORD_FILTER_OMIT_PRIVATE,
                                             &emsg))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "%s\n", emsg);
    GNUNET_free (emsg);
  }
}
 
 
static enum GNUNET_ErrorCode
store_record_set (struct NamestoreClient *nc,
                  const struct GNUNET_CRYPTO_BlindablePrivateKey *private_key,
                  const struct RecordSet *rd_set,
                  ssize_t *len)
{
  size_t name_len;
  size_t rd_ser_len;
  const char *name_tmp;
  const char *rd_ser;
  char *emsg;
  char *conv_name;
  unsigned int rd_count;
  int res;
  enum GNUNET_ErrorCode ec;
  struct GNUNET_TIME_Absolute new_block_exp;
  struct LookupExistingRecordsContext lctx;
  *len = sizeof (struct RecordSet);
 
  ec = GNUNET_EC_NONE;
  lctx.only_tombstone = GNUNET_NO;
  lctx.exp = GNUNET_TIME_UNIT_ZERO_ABS;
  new_block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
  name_len = ntohs (rd_set->name_len);
  *len += name_len;
  rd_count = ntohs (rd_set->rd_count);
  rd_ser_len = ntohs (rd_set->rd_len);
  *len += rd_ser_len;
  name_tmp = (const char *) &rd_set[1];
  rd_ser = &name_tmp[name_len];
  {
    struct GNUNET_GNSRECORD_Data rd[GNUNET_NZL (rd_count)];
 
    /* Extracting and converting private key */
    conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
    if (NULL == conv_name)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Error normalizing name `%s'\n",
                  name_tmp);
      return GNUNET_EC_NAMESTORE_LABEL_INVALID;
    }
 
    /* Check name for validity */
    if (GNUNET_OK != GNUNET_GNSRECORD_label_check (conv_name, &emsg))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Label invalid: `%s'\n",
                  emsg);
      GNUNET_free (emsg);
      GNUNET_free (conv_name);
      return GNUNET_EC_NAMESTORE_LABEL_INVALID;
    }
 
    if (GNUNET_OK !=
        GNUNET_GNSRECORD_records_deserialize (rd_ser_len, rd_ser, rd_count,
                                              rd))
    {
      GNUNET_free (conv_name);
      return GNUNET_EC_NAMESTORE_RECORD_DATA_INVALID;
    }
 
    GNUNET_STATISTICS_update (statistics,
                              "Well-formed store requests received",
                              1,
                              GNUNET_NO);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Creating %u records for name `%s'\n",
                (unsigned int) rd_count,
                conv_name);
    if ((0 == rd_count) &&
        (GNUNET_NO == nc->GSN_database->lookup_records (nc->GSN_database->cls,
                                                        private_key,
                                                        conv_name,
                                                        &get_existing_rd_exp,
                                                        &lctx)))
    {
      /* This name does not exist, so cannot be removed */
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Name `%s' does not exist, no deletion required\n",
                  conv_name);
      res = GNUNET_NO;
      ec = GNUNET_EC_NAMESTORE_RECORD_NOT_FOUND;
    }
    else
    {
      /* remove "NICK" records, unless this is for the
       #GNUNET_GNS_EMPTY_LABEL_AT label
       We may need one additional record later for tombstone.
       FIXME: Since we must normalize the record set (check for
       consistency etc) we have to iterate the set twice.
       May be inefficient.
       We cannot really move the nick caching into GNSRECORD.
       */
      struct GNUNET_GNSRECORD_Data rd_clean[GNUNET_NZL (rd_count)];
      struct GNUNET_GNSRECORD_Data rd_nf[GNUNET_NZL (rd_count) + 1];
      unsigned int rd_clean_off;
      unsigned int rd_nf_count;
      int have_nick;
 
      rd_clean_off = 0;
      have_nick = GNUNET_NO;
      for (unsigned int i = 0; i < rd_count; i++)
      {
        rd_clean[rd_clean_off] = rd[i];
 
        if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) ||
            (GNUNET_GNSRECORD_TYPE_NICK != rd[i].record_type))
          rd_clean_off++;
 
        if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) &&
            (GNUNET_GNSRECORD_TYPE_NICK == rd[i].record_type))
        {
          cache_nick (private_key, &rd[i]);
          have_nick = GNUNET_YES;
        }
      }
      if (GNUNET_OK !=
          GNUNET_GNSRECORD_normalize_record_set (conv_name,
                                                 rd_clean,
                                                 rd_clean_off,
                                                 rd_nf,
                                                 &rd_nf_count,
                                                 &new_block_exp,
                                                 GNUNET_GNSRECORD_FILTER_INCLUDE_MAINTENANCE,
                                                 &emsg))
      {
        GNUNET_free (conv_name);
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Error normalizing record set: `%s'\n",
                    emsg);
        GNUNET_free (emsg);
        return GNUNET_EC_NAMESTORE_RECORD_DATA_INVALID;
      }
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "%u/%u records before tombstone\n", rd_nf_count,
                  rd_clean_off);
      /*
       * If existing_block_exp is 0, then there was no record set
       * and no tombstone.
       * Otherwise, if the existing block expiration is after the
       * new block expiration would be, we need to add a tombstone
       * or update it.
       */
      if (GNUNET_TIME_absolute_cmp (new_block_exp, <=, lctx.exp))
      {
        rd_nf[rd_nf_count].record_type = GNUNET_GNSRECORD_TYPE_TOMBSTONE;
        rd_nf[rd_nf_count].expiration_time =
          lctx.exp.abs_value_us;
        rd_nf[rd_nf_count].data = NULL;
        rd_nf[rd_nf_count].data_size = 0;
        rd_nf[rd_nf_count].flags = GNUNET_GNSRECORD_RF_PRIVATE
                                   | GNUNET_GNSRECORD_RF_MAINTENANCE;
        rd_nf_count++;
      }
      if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) &&
          (GNUNET_NO == have_nick))
      {
        cache_nick (private_key, NULL);
      }
      res = nc->GSN_database->store_records (nc->GSN_database->cls,
                                             private_key,
                                             conv_name,
                                             rd_nf_count,
                                             rd_nf);
      /* If after a store there is only a TOMBSTONE left, and
       * there was >1 record under this label found (the tombstone; indicated
       * through res != GNUNET_NO) then we should return "NOT FOUND" == GNUNET_NO
       */
      if ((GNUNET_SYSERR != res) &&
          (0 == rd_count) &&
          (1 == rd_nf_count) &&
          (GNUNET_GNSRECORD_TYPE_TOMBSTONE == rd_nf[0].record_type) &&
          (GNUNET_YES == lctx.only_tombstone))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "Client tried to remove non-existent record\n");
        ec = GNUNET_EC_NAMESTORE_RECORD_NOT_FOUND;
      }
    }
 
    if (GNUNET_SYSERR == res)
    {
      /* store not successful, no need to tell monitors */
      GNUNET_free (conv_name);
      return GNUNET_EC_NAMESTORE_STORE_FAILED;
    }
  }
  return ec;
}
 
 
static void
handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  uint32_t rid;
  uint16_t rd_set_count;
  const char *buf;
  ssize_t read;
  size_t key_len;
  size_t kb_read;
  size_t rp_msg_len;
  size_t rs_len;
  size_t rs_off;
  struct StoreActivity *sa;
  struct RecordSet *rs;
  enum GNUNET_ErrorCode res;
 
  key_len = ntohs (rp_msg->key_len);
  rp_msg_len = ntohs (rp_msg->gns_header.header.size);
  rs_off = sizeof (*rp_msg) + key_len;
  rs_len = rp_msg_len - rs_off;
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_private_key_from_buffer (&rp_msg[1],
                                                   key_len,
                                                   &zone,
                                                   &kb_read)) ||
      (kb_read != key_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error reading private key\n");
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received NAMESTORE_RECORD_STORE message\n");
  rid = ntohl (rp_msg->gns_header.r_id);
  rd_set_count = ntohs (rp_msg->rd_set_count);
  buf = (const char *) rp_msg + rs_off;
  if (GNUNET_YES == ntohs (rp_msg->single_tx))
    nc->GSN_database->begin_tx (nc->GSN_database->cls);
  for (int i = 0; i < rd_set_count; i++)
  {
    rs = (struct RecordSet *) buf;
    res = store_record_set (nc, &zone,
                            rs, &read);
    if (GNUNET_EC_NONE != res)
    {
      if (GNUNET_YES == ntohs (rp_msg->single_tx))
        nc->GSN_database->rollback_tx (nc->GSN_database->cls);
      send_store_response (nc, res, rid);
      GNUNET_SERVICE_client_continue (nc->client);
      return;
    }
    buf += read;
  }
  if (GNUNET_YES == ntohs (rp_msg->single_tx))
    nc->GSN_database->commit_tx (nc->GSN_database->cls);
  sa = GNUNET_malloc (sizeof(struct StoreActivity) + rs_len);
  GNUNET_CONTAINER_DLL_insert (sa_head, sa_tail, sa);
  sa->nc = nc;
  sa->rs = (struct RecordSet *) &sa[1];
  sa->rd_set_count = rd_set_count;
  GNUNET_memcpy (&sa[1], (char *) rp_msg + rs_off, rs_len);
  sa->rid = rid;
  sa->rd_set_pos = 0;
  sa->private_key = zone;
  sa->zm_pos = monitor_head;
  continue_store_activity (sa, GNUNET_YES);
}
 
 
struct ZoneToNameCtx
{
  struct NamestoreClient *nc;
 
  uint32_t rid;
 
  enum GNUNET_ErrorCode ec;
};
 
 
static void
handle_zone_to_name_it (void *cls,
                        uint64_t seq,
                        const char *editor_hint,
                        const struct GNUNET_CRYPTO_BlindablePrivateKey *zone_key
                        ,
                        const char *name,
                        unsigned int rd_count,
                        const struct GNUNET_GNSRECORD_Data *rd)
{
  struct ZoneToNameCtx *ztn_ctx = cls;
  struct GNUNET_MQ_Envelope *env;
  struct ZoneToNameResponseMessage *ztnr_msg;
  size_t name_len;
  size_t key_len;
  ssize_t rd_ser_len;
  size_t msg_size;
  char *name_tmp;
  char *rd_tmp;
 
  GNUNET_assert (0 != seq);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Found result for zone-to-name lookup: `%s'\n",
              name);
  ztn_ctx->ec = GNUNET_EC_NONE;
  name_len = (NULL == name) ? 0 : strlen (name) + 1;
  rd_ser_len = GNUNET_GNSRECORD_records_get_size (rd_count, rd);
  if (rd_ser_len < 0)
  {
    GNUNET_break (0);
    ztn_ctx->ec = htonl (GNUNET_EC_NAMESTORE_UNKNOWN);
    return;
  }
  key_len = GNUNET_CRYPTO_blindable_sk_get_length (zone_key);
  msg_size = sizeof(struct ZoneToNameResponseMessage)
             + name_len + rd_ser_len + key_len;
  if (msg_size >= GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    ztn_ctx->ec = GNUNET_EC_NAMESTORE_RECORD_TOO_BIG;
    return;
  }
  env =
    GNUNET_MQ_msg_extra (ztnr_msg,
                         key_len + name_len + rd_ser_len,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_TO_NAME_RESPONSE);
  ztnr_msg->gns_header.header.size = htons (msg_size);
  ztnr_msg->gns_header.r_id = htonl (ztn_ctx->rid);
  ztnr_msg->ec = htonl (ztn_ctx->ec);
  ztnr_msg->rd_len = htons (rd_ser_len);
  ztnr_msg->rd_count = htons (rd_count);
  ztnr_msg->name_len = htons (name_len);
  ztnr_msg->key_len = htons (key_len);
  GNUNET_CRYPTO_write_blindable_sk_to_buffer (zone_key,
                                              &ztnr_msg[1],
                                              key_len);
  name_tmp = (char *) &ztnr_msg[1] + key_len;
  GNUNET_memcpy (name_tmp, name, name_len);
  rd_tmp = &name_tmp[name_len];
  GNUNET_assert (
    rd_ser_len ==
    GNUNET_GNSRECORD_records_serialize (rd_count, rd, rd_ser_len, rd_tmp));
  ztn_ctx->ec = GNUNET_EC_NONE;
  GNUNET_MQ_send (ztn_ctx->nc->mq, env);
}
 
 
static enum GNUNET_GenericReturnValue
check_zone_to_name (void *cls,
                    const struct ZoneToNameMessage *zis_msg)
{
  return GNUNET_OK;
}
 
 
static void
handle_zone_to_name (void *cls, const struct ZoneToNameMessage *ztn_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct GNUNET_CRYPTO_BlindablePublicKey value_zone;
  struct NamestoreClient *nc = cls;
  struct ZoneToNameCtx ztn_ctx;
  struct GNUNET_MQ_Envelope *env;
  struct ZoneToNameResponseMessage *ztnr_msg;
  size_t key_len;
  size_t pkey_len;
  size_t kb_read;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received ZONE_TO_NAME message\n");
  ztn_ctx.rid = ntohl (ztn_msg->gns_header.r_id);
  ztn_ctx.nc = nc;
  ztn_ctx.ec = GNUNET_EC_NAMESTORE_ZONE_NOT_FOUND;
  key_len = ntohs (ztn_msg->key_len);
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_private_key_from_buffer (&ztn_msg[1],
                                                   key_len,
                                                   &zone,
                                                   &kb_read)) ||
      (kb_read != key_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Error parsing private key.\n");
    GNUNET_SERVICE_client_drop (nc->client);
    GNUNET_break (0);
    return;
  }
  pkey_len = ntohs (ztn_msg->pkey_len);
  if ((GNUNET_SYSERR ==
       GNUNET_CRYPTO_read_blindable_pk_from_buffer ((char*) &ztn_msg[1]
                                                    + key_len,
                                                    pkey_len,
                                                    &value_zone,
                                                    &kb_read)) ||
      (kb_read != pkey_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Error parsing public key.\n");
    GNUNET_SERVICE_client_drop (nc->client);
    GNUNET_break (0);
    return;
  }
  if (GNUNET_SYSERR == nc->GSN_database->zone_to_name (nc->GSN_database->cls,
                                                       &zone,
                                                       &value_zone,
                                                       &handle_zone_to_name_it,
                                                       &ztn_ctx))
  {
    /* internal error, hang up instead of signalling something
       that might be wrong */
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  if (GNUNET_EC_NAMESTORE_ZONE_NOT_FOUND == ztn_ctx.ec)
  {
    /* no result found, send empty response */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Found no result for zone-to-name lookup.\n");
    env = GNUNET_MQ_msg (ztnr_msg,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_TO_NAME_RESPONSE);
    ztnr_msg->gns_header.r_id = ztn_msg->gns_header.r_id;
    ztnr_msg->ec = htonl (GNUNET_EC_NAMESTORE_ZONE_NOT_FOUND);
    GNUNET_MQ_send (nc->mq, env);
  }
  GNUNET_SERVICE_client_continue (nc->client);
}
 
 
struct ZoneIterationProcResult
{
  struct ZoneIteration *zi;
 
  uint64_t limit;
 
  int run_again;
};
 
 
static void
zone_iterate_proc (void *cls,
                   uint64_t seq,
                   const char *editor_hint,
                   const struct GNUNET_CRYPTO_BlindablePrivateKey *zone_key,
                   const char *name,
                   unsigned int rd_count,
                   const struct GNUNET_GNSRECORD_Data *rd)
{
  struct ZoneIterationProcResult *proc = cls;
 
  GNUNET_assert (0 != seq);
  if ((NULL == zone_key) && (NULL == name))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Iteration done\n");
    return;
  }
  if ((NULL == zone_key) || (NULL == name))
  {
    /* what is this!? should never happen */
    GNUNET_break (0);
    return;
  }
  if (0 == proc->limit)
  {
    /* what is this!? should never happen */
    GNUNET_break (0);
    return;
  }
  proc->zi->seq = seq;
  if (0 < send_lookup_response_with_filter (proc->zi->nc,
                                            proc->zi->request_id,
                                            zone_key,
                                            name,
                                            rd_count,
                                            rd,
                                            proc->zi->filter))
    proc->limit--;
  else
    proc->run_again = GNUNET_YES;
}
 
 
static void
run_zone_iteration_round (struct ZoneIteration *zi, uint64_t limit)
{
  struct ZoneIterationProcResult proc;
  struct GNUNET_TIME_Absolute start;
  struct GNUNET_TIME_Relative duration;
  struct NamestoreClient *nc = zi->nc;
 
  memset (&proc, 0, sizeof(proc));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Asked to return up to %llu records at position %llu\n",
              (unsigned long long) limit,
              (unsigned long long) zi->seq);
  proc.zi = zi;
  proc.limit = limit;
  proc.run_again = GNUNET_YES;
  start = GNUNET_TIME_absolute_get ();
  while (GNUNET_YES == proc.run_again)
  {
    proc.run_again = GNUNET_NO;
    GNUNET_break (GNUNET_SYSERR !=
                  nc->GSN_database->iterate_records (nc->GSN_database->cls,
                                                     (GNUNET_YES ==
                                                      GNUNET_is_zero (
                                                        &zi->zone))
                                               ? NULL
                                               : &zi->zone,
                                                     zi->seq,
                                                     proc.limit,
                                                     &zone_iterate_proc,
                                                     &proc));
  }
  duration = GNUNET_TIME_absolute_get_duration (start);
  duration = GNUNET_TIME_relative_divide (duration, limit - proc.limit);
  GNUNET_STATISTICS_set (statistics,
                         "NAMESTORE iteration delay (μs/record)",
                         duration.rel_value_us,
                         GNUNET_NO);
  if (0 == proc.limit)
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Returned %llu results, more results available\n",
                (unsigned long long) limit);
  zi->send_end = (0 != proc.limit);
  zone_iteration_done_client_continue (zi);
}
 
 
static enum GNUNET_GenericReturnValue
check_iteration_start (void *cls,
                       const struct ZoneIterationStartMessage *zis_msg)
{
  uint16_t size;
  size_t key_len;
 
  size = ntohs (zis_msg->gns_header.header.size);
  key_len = ntohs (zis_msg->key_len);
 
  if (size < key_len + sizeof(*zis_msg))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_iteration_start (void *cls,
                        const struct ZoneIterationStartMessage *zis_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  struct ZoneIteration *zi;
  size_t key_len;
  size_t kb_read;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ZONE_ITERATION_START message\n");
  key_len = ntohs (zis_msg->key_len);
  zi = GNUNET_new (struct ZoneIteration);
  if (0 < key_len)
  {
    if ((GNUNET_SYSERR ==
         GNUNET_CRYPTO_read_private_key_from_buffer (&zis_msg[1],
                                                     key_len,
                                                     &zone,
                                                     &kb_read)) ||
        (kb_read != key_len))
    {
      GNUNET_SERVICE_client_drop (nc->client);
      GNUNET_free (zi);
      return;
    }
    zi->zone = zone;
  }
  zi->request_id = ntohl (zis_msg->gns_header.r_id);
  zi->filter = ntohs (zis_msg->filter);
  zi->offset = 0;
  zi->nc = nc;
  GNUNET_CONTAINER_DLL_insert (nc->op_head, nc->op_tail, zi);
  run_zone_iteration_round (zi, 1);
}
 
 
static void
handle_iteration_stop (void *cls,
                       const struct ZoneIterationStopMessage *zis_msg)
{
  struct NamestoreClient *nc = cls;
  struct ZoneIteration *zi;
  uint32_t rid;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ZONE_ITERATION_STOP message\n");
  rid = ntohl (zis_msg->gns_header.r_id);
  for (zi = nc->op_head; NULL != zi; zi = zi->next)
    if (zi->request_id == rid)
      break;
  if (NULL == zi)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  GNUNET_CONTAINER_DLL_remove (nc->op_head, nc->op_tail, zi);
  GNUNET_free (zi);
  GNUNET_SERVICE_client_continue (nc->client);
}
 
 
static void
handle_iteration_next (void *cls,
                       const struct ZoneIterationNextMessage *zis_msg)
{
  struct NamestoreClient *nc = cls;
  struct ZoneIteration *zi;
  uint32_t rid;
  uint64_t limit;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ZONE_ITERATION_NEXT message\n");
  GNUNET_STATISTICS_update (statistics,
                            "Iteration NEXT messages received",
                            1,
                            GNUNET_NO);
  rid = ntohl (zis_msg->gns_header.r_id);
  limit = GNUNET_ntohll (zis_msg->limit);
  for (zi = nc->op_head; NULL != zi; zi = zi->next)
    if (zi->request_id == rid)
      break;
  if (NULL == zi)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  run_zone_iteration_round (zi, limit);
}
 
 
static void
monitor_unblock (struct ZoneMonitor *zm)
{
  struct StoreActivity *sa = sa_head;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Unblocking zone monitor %p\n", zm);
  while ((NULL != sa) && (zm->limit > zm->iteration_cnt))
  {
    struct StoreActivity *sn = sa->next;
 
    if (sa->zm_pos == zm)
      continue_store_activity (sa, GNUNET_YES);
    sa = sn;
  }
  if (zm->limit > zm->iteration_cnt)
  {
    zm->sa_waiting = GNUNET_NO;
    if (NULL != zm->sa_wait_warning)
    {
      GNUNET_SCHEDULER_cancel (zm->sa_wait_warning);
      zm->sa_wait_warning = NULL;
    }
  }
  else if (GNUNET_YES == zm->sa_waiting)
  {
    zm->sa_waiting_start = GNUNET_TIME_absolute_get ();
    if (NULL != zm->sa_wait_warning)
      GNUNET_SCHEDULER_cancel (zm->sa_wait_warning);
    zm->sa_wait_warning =
      GNUNET_SCHEDULER_add_delayed (MONITOR_STALL_WARN_DELAY,
                                    &warn_monitor_slow,
                                    zm);
  }
}
 
 
static void
monitor_sync (struct ZoneMonitor *zm)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_MessageHeader *sync;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Syncing zone monitor %p\n", zm);
 
  env = GNUNET_MQ_msg (sync, GNUNET_MESSAGE_TYPE_NAMESTORE_MONITOR_SYNC);
  GNUNET_MQ_send (zm->nc->mq, env);
  /* mark iteration done */
  zm->in_first_iteration = GNUNET_NO;
  zm->iteration_cnt = 0;
  if ((zm->limit > 0) && (zm->sa_waiting))
    monitor_unblock (zm);
}
 
 
static void
monitor_iteration_next (void *cls);
 
 
static void
monitor_iterate_cb (void *cls,
                    uint64_t seq,
                    const char *editor_hint,
                    const struct GNUNET_CRYPTO_BlindablePrivateKey *zone_key,
                    const char *name,
                    unsigned int rd_count,
                    const struct GNUNET_GNSRECORD_Data *rd)
{
  struct ZoneMonitor *zm = cls;
 
  GNUNET_assert (0 != seq);
  zm->seq = seq;
  GNUNET_assert (NULL != name);
  GNUNET_STATISTICS_update (statistics,
                            "Monitor notifications sent",
                            1,
                            GNUNET_NO);
  if (0 < send_lookup_response_with_filter (zm->nc, 0, zone_key, name,
                                            rd_count, rd, zm->filter))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sent records.\n");
    zm->limit--;
    zm->iteration_cnt--;
  }
  else
    zm->run_again = GNUNET_YES;
  if ((0 == zm->iteration_cnt) && (0 != zm->limit))
  {
    /* We are done with the current iteration batch, AND the
       client would right now accept more, so go again! */
    GNUNET_assert (NULL == zm->task);
    zm->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next, zm);
  }
}
 
 
static enum GNUNET_GenericReturnValue
check_monitor_start (void *cls,
                     const struct ZoneMonitorStartMessage *zis_msg)
{
  uint16_t size;
  size_t key_len;
 
  size = ntohs (zis_msg->header.size);
  key_len = ntohs (zis_msg->key_len);
 
  if (size < key_len + sizeof(*zis_msg))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_monitor_start (void *cls, const struct
                      ZoneMonitorStartMessage *zis_msg)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
  struct NamestoreClient *nc = cls;
  struct ZoneMonitor *zm;
  size_t key_len;
  size_t kb_read;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ZONE_MONITOR_START message\n");
  zm = GNUNET_new (struct ZoneMonitor);
  zm->nc = nc;
  key_len = ntohs (zis_msg->key_len);
  if (0 < key_len)
  {
    if ((GNUNET_SYSERR ==
         GNUNET_CRYPTO_read_private_key_from_buffer (&zis_msg[1],
                                                     key_len,
                                                     &zone,
                                                     &kb_read)) ||
        (kb_read != key_len))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Error reading private key\n");
      GNUNET_SERVICE_client_drop (nc->client);
      GNUNET_free (zm);
      return;
    }
    zm->zone = zone;
  }
  zm->limit = 1;
  zm->filter = ntohs (zis_msg->filter);
  zm->in_first_iteration = (GNUNET_YES == ntohl (zis_msg->iterate_first));
  GNUNET_CONTAINER_DLL_insert (monitor_head, monitor_tail, zm);
  GNUNET_SERVICE_client_mark_monitor (nc->client);
  GNUNET_SERVICE_client_continue (nc->client);
  GNUNET_notification_context_add (monitor_nc, nc->mq);
  if (zm->in_first_iteration)
    zm->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next, zm);
  else
    monitor_sync (zm);
}
 
 
static void
monitor_iteration_next (void *cls)
{
  struct ZoneMonitor *zm = cls;
  struct NamestoreClient *nc = zm->nc;
  int ret;
 
  zm->task = NULL;
  GNUNET_assert (0 == zm->iteration_cnt);
  if (zm->limit > 16)
    zm->iteration_cnt = zm->limit / 2;   /* leave half for monitor events */
  else
    zm->iteration_cnt = zm->limit;   /* use it all */
  zm->run_again = GNUNET_YES;
  while (GNUNET_YES == zm->run_again)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Running iteration\n");
    zm->run_again = GNUNET_NO;
    ret = nc->GSN_database->iterate_records (nc->GSN_database->cls,
                                             (GNUNET_YES == GNUNET_is_zero (
                                                &zm->zone)) ? NULL : &zm->zone,
                                             zm->seq,
                                             zm->iteration_cnt,
                                             &monitor_iterate_cb,
                                             zm);
  }
  if (GNUNET_SYSERR == ret)
  {
    GNUNET_SERVICE_client_drop (zm->nc->client);
    return;
  }
  if (GNUNET_NO == ret)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Zone empty... syncing\n");
    /* empty zone */
    monitor_sync (zm);
    return;
  }
}
 
 
static void
handle_monitor_next (void *cls, const struct ZoneMonitorNextMessage *nm)
{
  struct NamestoreClient *nc = cls;
  struct ZoneMonitor *zm;
  uint64_t inc;
 
  inc = GNUNET_ntohll (nm->limit);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ZONE_MONITOR_NEXT message with limit %llu\n",
              (unsigned long long) inc);
  for (zm = monitor_head; NULL != zm; zm = zm->next)
    if (zm->nc == nc)
      break;
  if (NULL == zm)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  GNUNET_SERVICE_client_continue (nc->client);
  if (zm->limit + inc < zm->limit)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (nc->client);
    return;
  }
  zm->limit += inc;
  if ((zm->in_first_iteration) && (zm->limit == inc))
  {
    /* We are still iterating, and the previous iteration must
       have stopped due to the client's limit, so continue it! */
    GNUNET_assert (NULL == zm->task);
    zm->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next, zm);
  }
  GNUNET_assert (zm->iteration_cnt <= zm->limit);
  if ((zm->limit > zm->iteration_cnt) && (zm->sa_waiting))
  {
    monitor_unblock (zm);
  }
  else if (GNUNET_YES == zm->sa_waiting)
  {
    if (NULL != zm->sa_wait_warning)
      GNUNET_SCHEDULER_cancel (zm->sa_wait_warning);
    zm->sa_waiting_start = GNUNET_TIME_absolute_get ();
    zm->sa_wait_warning =
      GNUNET_SCHEDULER_add_delayed (MONITOR_STALL_WARN_DELAY,
                                    &warn_monitor_slow,
                                    zm);
  }
}
 
 
static void
run (void *cls,
     const struct GNUNET_CONFIGURATION_Handle *cfg,
     struct GNUNET_SERVICE_Handle *service)
{
  char *database;
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Starting namestore service\n");
  return_orphaned = GNUNET_CONFIGURATION_get_value_yesno (cfg,
                                                          "namestore",
                                                          "RETURN_ORPHANED");
  GSN_cfg = cfg;
  monitor_nc = GNUNET_notification_context_create (1);
  statistics = GNUNET_STATISTICS_create ("namestore", cfg);
  /* Loading database plugin */
  if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_string (cfg,
                                                          "namestore",
                                                          "database",
                                                          &database))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "No database backend configured\n");
    GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
    return;
  }
  GNUNET_asprintf (&db_lib_name,
                   "libgnunet_plugin_namestore_%s",
                   database);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Loading %s\n",
              db_lib_name);
  GSN_database = GNUNET_PLUGIN_load (GNUNET_OS_project_data_gnunet (),
                                     db_lib_name,
                                     (void *) cfg);
  GNUNET_free (database);
  if (NULL == GSN_database)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Could not load database backend `%s'\n",
                db_lib_name);
    GNUNET_free (db_lib_name);
    GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
    return;
  }
  GNUNET_SCHEDULER_add_shutdown (&cleanup_task, NULL);
}
 
 
GNUNET_SERVICE_MAIN (
  GNUNET_OS_project_data_gnunet (),
  "namestore",
  GNUNET_SERVICE_OPTION_NONE,
  &run,
  &client_connect_cb,
  &client_disconnect_cb,
  NULL,
  GNUNET_MQ_hd_var_size (record_store,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE,
                         struct RecordStoreMessage,
                         NULL),
  GNUNET_MQ_hd_var_size (edit_record_set,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_SET_EDIT,
                         struct EditRecordSetMessage,
                         NULL),
  GNUNET_MQ_hd_var_size (edit_record_set_cancel,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_SET_EDIT_CANCEL,
                         struct EditRecordSetCancelMessage,
                         NULL),
  GNUNET_MQ_hd_var_size (record_lookup,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP,
                         struct LabelLookupMessage,
                         NULL),
  GNUNET_MQ_hd_var_size (zone_to_name,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_TO_NAME,
                         struct ZoneToNameMessage,
                         NULL),
  GNUNET_MQ_hd_var_size (iteration_start,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_ITERATION_START,
                         struct ZoneIterationStartMessage,
                         NULL),
  GNUNET_MQ_hd_fixed_size (iteration_next,
                           GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_ITERATION_NEXT,
                           struct ZoneIterationNextMessage,
                           NULL),
  GNUNET_MQ_hd_fixed_size (iteration_stop,
                           GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_ITERATION_STOP,
                           struct ZoneIterationStopMessage,
                           NULL),
  GNUNET_MQ_hd_var_size (monitor_start,
                         GNUNET_MESSAGE_TYPE_NAMESTORE_MONITOR_START,
                         struct ZoneMonitorStartMessage,
                         NULL),
  GNUNET_MQ_hd_fixed_size (monitor_next,
                           GNUNET_MESSAGE_TYPE_NAMESTORE_MONITOR_NEXT,
                           struct ZoneMonitorNextMessage,
                           NULL),
  GNUNET_MQ_handler_end ());
 
 
/* end of gnunet-service-namestore.c */