NAMESTORE: Bulk store API and fix for delayed store activities

New API: GNUNET_NAMESTORE_records_store2 which allows the caller
to pass an array of records in order to facilitate bulk import of zone
data.

Further, the transactional API requires that monitors and namecache
updates are delayed until transactions are actually commited.

11 files changed, 440 insertions, 158 deletions

diff --git a/src/include/gnunet_namestore_service.h b/src/include/gnunet_namestore_service.h
index dc6aeaa69..b795494af 100644
--- a/src/include/gnunet_namestore_service.h
+++ b/src/include/gnunet_namestore_service.h
@@ -148,6 +148,43 @@ GNUNET_NAMESTORE_records_store (struct GNUNET_NAMESTORE_Handle *h,
                                 void *cont_cls);
 
 /**
+ * Store one or more record sets in the namestore. If any item is already present,
+ * it is replaced with the new record.  Use an empty array to
+ * remove all records under the given name.
+ *
+ * The continuation is called after the records have been stored in the
+ * database. They may not yet have been commited. Monitors may be notified
+ * asynchronously (basically with a buffer when commited).
+ * However, if any monitor is consistently too slow to
+ * keep up with the changes, calling @a cont will be delayed until the
+ * monitors do keep up.
+ * Uncommited store requests within a transaction (GNUNET_NAMESTORE_transaction_begin)
+ * cause @a cont to be called immediately before the commit and before
+ * notification of monitors.
+ *
+ * @param h handle to the namestore
+ * @param pkey private key of the zone
+ * @param rd_set_count the number of record sets
+ * @param a_label an array of size @a rd_set_count of names for each record set
+ * @param a_rd_count an array of size @a rd_set_count containing the number of records in the corresponding 'rd' array
+ * @param a_rd an array of size @a rd_set_count of arrays of records with data to store
+ * @param cont continuation to call when done
+ * @param cont_cls closure for @a cont
+ * @return handle to abort the request
+ */
+
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_store2 (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  unsigned int rd_set_count,
+  const char **a_label,
+  unsigned int *a_rd_count,
+  const struct GNUNET_GNSRECORD_Data **a_rd,
+  GNUNET_NAMESTORE_ContinuationWithStatus cont,
+  void *cont_cls);
+
+/**
  * Store an item in the namestore.  If the item is already present,
  * it is replaced with the new record.  Use an empty array to
  * remove all records under the given name.
diff --git a/src/namestore/gnunet-service-namestore.c b/src/namestore/gnunet-service-namestore.c
index b860d7a3c..ada2acb16 100644
--- a/src/namestore/gnunet-service-namestore.c
+++ b/src/namestore/gnunet-service-namestore.c
@@ -147,6 +147,11 @@ struct NamestoreClient
   char *db_lib_name;
 
   /**
+   * GNUNET_YES if this nc has begun a transaction which is uncommited.
+   */
+  int in_transaction;
+
+  /**
    * Message queue for transmission to @e client
    */
   struct GNUNET_MQ_Handle *mq;
@@ -308,10 +313,26 @@ struct StoreActivity
   struct NamestoreClient *nc;
 
   /**
-   * Copy of the original store message (as data fields in @e rd will
+   * The request ID
+   */
+  uint32_t rid;
+
+  /**
+   * Wheather or not this store action is already commited.
+   * The store activity will not be processed unless this field is GNUNET_YES
+   */
+  int uncommited;
+
+  /**
+   * The zone private key
+   */
+  struct GNUNET_IDENTITY_PrivateKey private_key;
+
+  /**
+   * Copy of the original record set (as data fields in @e rd will
    * point into it!).
    */
-  const struct RecordStoreMessage *rsm;
+  const struct RecordSet *rs;
 
   /**
    * Next zone monitor that still needs to be notified about this PUT.
@@ -879,7 +900,7 @@ send_lookup_response (struct NamestoreClient *nc,
  * @param rid client's request ID
  */
 static void
-send_store_response (struct NamestoreClient *nc, int res, const char*emsg,
+send_store_response (struct NamestoreClient *nc, int res, const char *emsg,
                      uint32_t rid)
 {
   struct GNUNET_MQ_Envelope *env;
@@ -1105,7 +1126,7 @@ warn_monitor_slow (void *cls)
 static void
 continue_store_activity (struct StoreActivity *sa)
 {
-  const struct RecordStoreMessage *rp_msg = sa->rsm;
+  const struct RecordSet *rd_set = sa->rs;
   unsigned int rd_count;
   size_t name_len;
   size_t rd_ser_len;
@@ -1113,11 +1134,20 @@ continue_store_activity (struct StoreActivity *sa)
   const char *name_tmp;
   const char *rd_ser;
 
-  rid = ntohl (rp_msg->gns_header.r_id);
-  name_len = ntohs (rp_msg->name_len);
-  rd_count = ntohs (rp_msg->rd_count);
-  rd_ser_len = ntohs (rp_msg->rd_len);
-  name_tmp = (const char *) &rp_msg[1];
+  // If we are in a transaction, do not notify monitors or update
+  // cached. This will be done when we are commiting.
+  if (GNUNET_YES == sa->uncommited)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Transaction not yet committed, delaying monitor and cache updates\n");
+    send_store_response (sa->nc, GNUNET_YES, NULL, sa->rid);
+    GNUNET_SERVICE_client_continue (sa->nc->client);
+    return;
+  }
+  name_len = ntohs (rd_set->name_len);
+  rd_count = ntohs (rd_set->rd_count);
+  rd_ser_len = ntohs (rd_set->rd_len);
+  name_tmp = (const char *) &rd_set[1];
   rd_ser = &name_tmp[name_len];
   {
     struct GNUNET_GNSRECORD_Data rd[GNUNET_NZL (rd_count)];
@@ -1129,7 +1159,7 @@ continue_store_activity (struct StoreActivity *sa)
 
     for (struct ZoneMonitor *zm = sa->zm_pos; NULL != zm; zm = sa->zm_pos)
     {
-      if ((0 != GNUNET_memcmp (&rp_msg->private_key, &zm->zone)) &&
+      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 */
@@ -1153,7 +1183,7 @@ continue_store_activity (struct StoreActivity *sa)
       zm->limit--;
       send_lookup_response_with_filter (zm->nc,
                                         0,
-                                        &rp_msg->private_key,
+                                        &sa->private_key,
                                         sa->conv_name,
                                         rd_count,
                                         rd,
@@ -1163,8 +1193,8 @@ continue_store_activity (struct StoreActivity *sa)
     /* great, done with the monitors, unpack (again) for refresh_block operation */
     refresh_block (sa->nc,
                    NULL,
-                   rid,
-                   &rp_msg->private_key,
+                   sa->rid,
+                   &sa->private_key,
                    sa->conv_name,
                    rd_count,
                    rd);
@@ -1240,7 +1270,8 @@ client_disconnect_cb (void *cls,
   for (cop = cop_head; NULL != cop; cop = cop->next)
     if (nc == cop->nc)
       cop->nc = NULL;
-  GNUNET_break (NULL == GNUNET_PLUGIN_unload (nc->db_lib_name, nc->GSN_database));
+  GNUNET_break (NULL == GNUNET_PLUGIN_unload (nc->db_lib_name,
+                                              nc->GSN_database));
   GNUNET_free (nc->db_lib_name);
   GNUNET_free (nc);
 }
@@ -1549,29 +1580,16 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
 static int
 check_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
 {
-  size_t name_len;
   size_t msg_size;
-  size_t msg_size_exp;
-  size_t rd_ser_len;
-  const char *name_tmp;
+  size_t min_size_exp;
+  size_t rd_set_count;
 
   (void) cls;
-  name_len = ntohs (rp_msg->name_len);
   msg_size = ntohs (rp_msg->gns_header.header.size);
-  rd_ser_len = ntohs (rp_msg->rd_len);
-  msg_size_exp = sizeof(struct RecordStoreMessage) + name_len + rd_ser_len;
-  if (msg_size != msg_size_exp)
-  {
-    GNUNET_break (0);
-    return GNUNET_SYSERR;
-  }
-  if ((0 == name_len) || (name_len > MAX_NAME_LEN))
-  {
-    GNUNET_break (0);
-    return GNUNET_SYSERR;
-  }
-  name_tmp = (const char *) &rp_msg[1];
-  if ('\0' != name_tmp[name_len - 1])
+  rd_set_count = ntohs (rp_msg->rd_set_count);
+  min_size_exp = sizeof(struct RecordStoreMessage) + sizeof (struct RecordSet)
+                 * rd_set_count;
+  if (msg_size < min_size_exp)
   {
     GNUNET_break (0);
     return GNUNET_SYSERR;
@@ -1620,77 +1638,61 @@ get_block_exp_existing (void *cls,
   }
 }
 
-
-/**
- * Handles a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE message
- *
- * @param cls client sending the message
- * @param rp_msg message of type `struct RecordStoreMessage`
- */
-static void
-handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
+static enum GNUNET_GenericReturnValue
+store_record_set (struct NamestoreClient *nc,
+                  const struct GNUNET_IDENTITY_PrivateKey *private_key,
+                  const struct RecordSet *rd_set,
+                  char **conv_name,
+                  ssize_t *len,
+                  char **emsg)
 {
-  struct NamestoreClient *nc = cls;
   size_t name_len;
   size_t rd_ser_len;
-  uint32_t rid;
   const char *name_tmp;
-  char *conv_name;
   const char *rd_ser;
   unsigned int rd_count;
   int res;
-  struct StoreActivity *sa;
   struct GNUNET_TIME_Absolute existing_block_exp;
   struct GNUNET_TIME_Absolute new_block_exp;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Received NAMESTORE_RECORD_STORE message\n");
-  existing_block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
-  new_block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
-  rid = ntohl (rp_msg->gns_header.r_id);
-  name_len = ntohs (rp_msg->name_len);
-  rd_count = ntohs (rp_msg->rd_count);
-  rd_ser_len = ntohs (rp_msg->rd_len);
-  name_tmp = (const char *) &rp_msg[1];
+  *len = sizeof (struct RecordSet);
+
+  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)];
-    char *emsg;
 
     /* Extracting and converting private key */
-    conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
-    if (NULL == conv_name)
+    *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);
-      send_store_response (nc, GNUNET_SYSERR, _ ("Error normalizing name."),
-                           rid);
-      GNUNET_SERVICE_client_continue (nc->client);
-      return;
+      *emsg = GNUNET_strdup (_ ("Error normalizing name."));
+      return GNUNET_SYSERR;
     }
 
     /* Check name for validity */
-    if (GNUNET_OK != GNUNET_GNSRECORD_label_check (conv_name, &emsg))
+    if (GNUNET_OK != GNUNET_GNSRECORD_label_check (*conv_name, emsg))
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                   "Label invalid: `%s'\n",
-                  emsg);
-      send_store_response (nc, GNUNET_SYSERR, emsg, rid);
-      GNUNET_free (emsg);
-      GNUNET_free (conv_name);
-      GNUNET_SERVICE_client_continue (nc->client);
-      return;
+                  *emsg);
+      GNUNET_free (*conv_name);
+      return -1;
     }
 
     if (GNUNET_OK !=
         GNUNET_GNSRECORD_records_deserialize (rd_ser_len, rd_ser, rd_count, rd))
     {
-      send_store_response (nc, GNUNET_SYSERR,
-                           _ ("Error deserializing records."), rid);
-      GNUNET_free (conv_name);
-      GNUNET_SERVICE_client_continue (nc->client);
-      return;
+      *emsg = GNUNET_strdup (_ ("Error deserializing records."));
+      GNUNET_free (*conv_name);
+      return GNUNET_SYSERR;
     }
 
     GNUNET_STATISTICS_update (statistics,
@@ -1700,10 +1702,10 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "Creating %u records for name `%s'\n",
                 (unsigned int) rd_count,
-                conv_name);
+                *conv_name);
     if ((GNUNET_NO == nc->GSN_database->lookup_records (nc->GSN_database->cls,
-                                                        &rp_msg->private_key,
-                                                        conv_name,
+                                                        private_key,
+                                                        *conv_name,
                                                         &get_block_exp_existing,
                                                         &existing_block_exp)) &&
         (rd_count == 0))
@@ -1711,7 +1713,7 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       /* 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);
+                  *conv_name);
       res = GNUNET_NO;
     }
     else
@@ -1728,7 +1730,6 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       struct GNUNET_GNSRECORD_Data rd_nf[GNUNET_NZL (rd_count) + 1];
       unsigned int rd_clean_off;
       unsigned int rd_nf_count;
-      char *emsg;
       int have_nick;
 
       rd_clean_off = 0;
@@ -1737,35 +1738,34 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       {
         rd_clean[rd_clean_off] = rd[i];
 
-        if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) ||
+        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)) &&
+        if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, *conv_name)) &&
             (GNUNET_GNSRECORD_TYPE_NICK == rd[i].record_type))
         {
-          cache_nick (&rp_msg->private_key, &rd[i]);
+          // FIXME: In case this is an uncommited transaction,
+          // we should not do this here. Can we do this in the store activity?
+          cache_nick (private_key, &rd[i]);
           have_nick = GNUNET_YES;
         }
       }
       if (GNUNET_OK !=
-          GNUNET_GNSRECORD_normalize_record_set (conv_name,
+          GNUNET_GNSRECORD_normalize_record_set (*conv_name,
                                                  rd_clean,
                                                  rd_clean_off,
                                                  rd_nf,
                                                  &rd_nf_count,
                                                  &new_block_exp,
                                                  GNUNET_GNSRECORD_FILTER_NONE,
-                                                 &emsg))
+                                                 emsg))
       {
-        send_store_response (nc, GNUNET_SYSERR, emsg, rid);
-        GNUNET_free (emsg);
-        GNUNET_SERVICE_client_continue (nc->client);
-        GNUNET_free (conv_name);
-        return;
+        GNUNET_free (*conv_name);
+        return GNUNET_SYSERR;
       }
       /*
-       * If existing_block_exp is 0, then there was not record set
+       * 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
@@ -1781,36 +1781,91 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
         rd_nf[rd_nf_count].flags = GNUNET_GNSRECORD_RF_PRIVATE;
         rd_nf_count++;
       }
-      if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) &&
+      if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, *conv_name)) &&
           (GNUNET_NO == have_nick))
       {
         /* remove nick record from cache, in case we have one there */
-        cache_nick (&rp_msg->private_key, NULL);
+        // FIXME: In case this is an uncommited transaction,
+        // we should not do this here. Can we do this in the store activity?
+        cache_nick (private_key, NULL);
       }
       res = nc->GSN_database->store_records (nc->GSN_database->cls,
-                                             &rp_msg->private_key,
-                                             conv_name,
+                                             private_key,
+                                             *conv_name,
                                              rd_nf_count,
                                              rd_nf);
     }
 
-    if (GNUNET_OK != res)
+    if (GNUNET_SYSERR == res)
     {
       /* store not successful, no need to tell monitors */
-      send_store_response (nc, res, _ ("Store failed"), rid);
+      *emsg = GNUNET_strdup (_ ("Store failed"));
+      GNUNET_free (*conv_name);
+      return GNUNET_SYSERR;
+    }
+  }
+  return res;
+}
+
+/**
+ * Handles a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE message
+ *
+ * @param cls client sending the message
+ * @param rp_msg message of type `struct RecordStoreMessage`
+ */
+static void
+handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
+{
+  struct NamestoreClient *nc = cls;
+  size_t name_len;
+  size_t rd_ser_len;
+  uint32_t rid;
+  uint16_t rd_set_count;
+  const char *name_tmp;
+  char *conv_name;
+  char *emsg = NULL;
+  const char *buf;
+  const char *rd_ser;
+  unsigned int rd_count;
+  ssize_t read;
+  struct StoreActivity *sa;
+  struct RecordSet *rs;
+  struct GNUNET_TIME_Absolute existing_block_exp;
+  struct GNUNET_TIME_Absolute new_block_exp;
+  enum GNUNET_GenericReturnValue res;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received NAMESTORE_RECORD_STORE message\n");
+  existing_block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
+  new_block_exp = GNUNET_TIME_UNIT_ZERO_ABS;
+  rid = ntohl (rp_msg->gns_header.r_id);
+  rd_set_count = ntohs (rp_msg->rd_set_count);
+  buf = (const char *) &rp_msg[1];
+  for (int i = 0; i < rd_set_count; i++)
+  {
+    rs = (struct RecordSet *) buf;
+    res = store_record_set (nc, &rp_msg->private_key,
+                            rs, &conv_name, &read, &emsg);
+    if (GNUNET_OK != res)
+    {
+      send_store_response (nc, res, emsg,
+                           rid);
+      GNUNET_free (emsg);
       GNUNET_SERVICE_client_continue (nc->client);
-      GNUNET_free (conv_name);
       return;
     }
-    sa = GNUNET_malloc (sizeof(struct StoreActivity)
-                        + ntohs (rp_msg->gns_header.header.size));
+    sa = GNUNET_malloc (sizeof(struct StoreActivity) + read);
     GNUNET_CONTAINER_DLL_insert (sa_head, sa_tail, sa);
     sa->nc = nc;
-    sa->rsm = (const struct RecordStoreMessage *) &sa[1];
-    GNUNET_memcpy (&sa[1], rp_msg, ntohs (rp_msg->gns_header.header.size));
+    sa->rs = (const struct RecordSet *) &sa[1];
+    GNUNET_memcpy (&sa[1], rs, read);
+    sa->rid = rid;
+    sa->private_key = rp_msg->private_key;
     sa->zm_pos = monitor_head;
     sa->conv_name = conv_name;
+    sa->uncommited = nc->in_transaction;
     continue_store_activity (sa);
+    buf += read;
   }
 }
 
@@ -1826,6 +1881,8 @@ handle_tx_control (void *cls, const struct TxControlMessage *tx_msg)
   struct NamestoreClient *nc = cls;
   struct TxControlResultMessage *txr_msg;
   struct GNUNET_MQ_Envelope *env;
+  struct StoreActivity *sa = sa_head;
+  struct StoreActivity *sn;
   enum GNUNET_GenericReturnValue ret;
   char *emsg = NULL;
   char *err_tmp;
@@ -1836,14 +1893,47 @@ handle_tx_control (void *cls, const struct TxControlMessage *tx_msg)
   case GNUNET_NAMESTORE_TX_BEGIN:
     ret = nc->GSN_database->transaction_begin (nc->GSN_database->cls,
                                                &emsg);
+    nc->in_transaction = GNUNET_YES;
     break;
   case GNUNET_NAMESTORE_TX_COMMIT:
     ret = nc->GSN_database->transaction_commit (nc->GSN_database->cls,
                                                 &emsg);
+    if (GNUNET_SYSERR != ret)
+    {
+      nc->in_transaction = GNUNET_NO;
+      while (NULL != sa)
+      {
+        if ((nc != sa->nc) ||
+            (GNUNET_NO == sa->uncommited))
+        {
+          sa = sa->next;
+          continue;
+        }
+        sa->uncommited = GNUNET_NO;
+        continue_store_activity (sa);
+        sa = sa->next;
+      }
+    }
     break;
   case GNUNET_NAMESTORE_TX_ROLLBACK:
     ret = nc->GSN_database->transaction_rollback (nc->GSN_database->cls,
                                                   &emsg);
+    if (GNUNET_SYSERR != ret)
+    {
+      nc->in_transaction = GNUNET_NO;
+      while (NULL != sa)
+      {
+        if ((nc != sa->nc) ||
+            (GNUNET_NO == sa->uncommited))
+        {
+          sa = sa->next;
+          continue;
+        }
+        sn = sa->next;
+        free_store_activity (sa);
+        sa = sn;
+      }
+    }
     break;
   default:
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
diff --git a/src/namestore/namestore.conf.in b/src/namestore/namestore.conf.in
index a9c928c66..fe7d4002c 100644
--- a/src/namestore/namestore.conf.in
+++ b/src/namestore/namestore.conf.in
@@ -21,12 +21,9 @@ CACHE_KEYS = YES
 
 
 [namestore-sqlite]
+INIT_ON_CONNECT = YES
 FILENAME = $GNUNET_DATA_HOME/namestore/sqlite.db
 
-[namestore-heap]
-FILENAME = $GNUNET_DATA_HOME/namestore/heap.db
-
-
 [namestore-postgres]
 # How to connect to the database
 CONFIG = postgres:///gnunet
@@ -34,6 +31,7 @@ CONFIG = postgres:///gnunet
 TEMPORARY_TABLE = NO
 # Use asynchronous commit (SET synchronous_commit TO OFF).
 ASYNC_COMMIT = NO
+INIT_ON_CONNECT = YES
 
 [uri]
 gns = gnunet-namestore -e 1a -u
diff --git a/src/namestore/namestore.h b/src/namestore/namestore.h
index ad02ffb48..d7b6fd13e 100644
--- a/src/namestore/namestore.h
+++ b/src/namestore/namestore.h
@@ -51,23 +51,9 @@ struct GNUNET_NAMESTORE_Header
   uint32_t r_id GNUNET_PACKED;
 };
 
-
-/**
- * Store a record to the namestore (as authority).
- */
-struct RecordStoreMessage
+struct RecordSet
 {
   /**
-   * Type will be #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE
-   */
-  struct GNUNET_NAMESTORE_Header gns_header;
-
-  /**
-   * Expiration time
-   */
-  struct GNUNET_TIME_AbsoluteNBO expire;
-
-  /**
    * Name length
    */
   uint16_t name_len GNUNET_PACKED;
@@ -87,14 +73,35 @@ struct RecordStoreMessage
    */
   uint16_t reserved GNUNET_PACKED;
 
+
+  /* followed by:
+   * name with length name_len
+   * serialized record data with rd_count records
+   */
+};
+
+/**
+ * Store a record to the namestore (as authority).
+ */
+struct RecordStoreMessage
+{
+  /**
+   * Type will be #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE
+   */
+  struct GNUNET_NAMESTORE_Header gns_header;
+
   /**
    * The private key of the authority.
    */
   struct GNUNET_IDENTITY_PrivateKey private_key;
 
-  /* followed by:
-   * name with length name_len
-   * serialized record data with rd_count records
+  /**
+   * Number of record sets
+   */
+  uint16_t rd_set_count;
+
+  /**
+   * Followed by rd_set_count RecordSets
    */
 };
 
diff --git a/src/namestore/namestore_api.c b/src/namestore/namestore_api.c
index 51ee25acf..f9ed2b620 100644
--- a/src/namestore/namestore_api.c
+++ b/src/namestore/namestore_api.c
@@ -1095,32 +1095,60 @@ GNUNET_NAMESTORE_records_store (
   GNUNET_NAMESTORE_ContinuationWithStatus cont,
   void *cont_cls)
 {
+  return GNUNET_NAMESTORE_records_store2 (h, pkey, 1, &label, &rd_count, &rd,
+                                          cont, cont_cls);
+}
+
+struct GNUNET_NAMESTORE_QueueEntry *
+GNUNET_NAMESTORE_records_store2 (
+  struct GNUNET_NAMESTORE_Handle *h,
+  const struct GNUNET_IDENTITY_PrivateKey *pkey,
+  unsigned int rd_set_count,
+  const char **a_label,
+  unsigned int *a_rd_count,
+  const struct GNUNET_GNSRECORD_Data **a_rd,
+  GNUNET_NAMESTORE_ContinuationWithStatus cont,
+  void *cont_cls)
+{
   struct GNUNET_NAMESTORE_QueueEntry *qe;
   struct GNUNET_MQ_Envelope *env;
+  const char *label;
+  unsigned int rd_count;
+  const struct GNUNET_GNSRECORD_Data *rd;
   char *name_tmp;
   char *rd_ser;
-  ssize_t rd_ser_len;
+  ssize_t rd_ser_len[rd_set_count];
   size_t name_len;
   uint32_t rid;
   struct RecordStoreMessage *msg;
+  struct RecordSet *rd_set;
   ssize_t sret;
+  int i;
+  size_t rd_set_len = 0;
 
-  name_len = strlen (label) + 1;
-  if (name_len > MAX_NAME_LEN)
+  for (i = 0; i < rd_set_count; i++)
   {
-    GNUNET_break (0);
-    return NULL;
-  }
-  rd_ser_len = GNUNET_GNSRECORD_records_get_size (rd_count, rd);
-  if (rd_ser_len < 0)
-  {
-    GNUNET_break (0);
-    return NULL;
-  }
-  if (rd_ser_len > UINT16_MAX)
-  {
-    GNUNET_break (0);
-    return NULL;
+    label = a_label[i];
+    rd_count = a_rd_count[i];
+    rd = a_rd[i];
+    name_len = strlen (label) + 1;
+    if (name_len > MAX_NAME_LEN)
+    {
+      GNUNET_break (0);
+      return NULL;
+    }
+    rd_ser_len[i] = GNUNET_GNSRECORD_records_get_size (rd_count, rd);
+    if (rd_ser_len[i] < 0)
+    {
+      GNUNET_break (0);
+      return NULL;
+    }
+    if (rd_ser_len[i] > UINT16_MAX)
+    {
+      GNUNET_break (0);
+      return NULL;
+    }
+    rd_set_len = sizeof (struct RecordSet) + name_len + rd_ser_len[i];
   }
   rid = get_op_id (h);
   qe = GNUNET_new (struct GNUNET_NAMESTORE_QueueEntry);
@@ -1132,30 +1160,38 @@ GNUNET_NAMESTORE_records_store (
 
   /* setup msg */
   env = GNUNET_MQ_msg_extra (msg,
-                             name_len + rd_ser_len,
+                             rd_set_len,
                              GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE);
   msg->gns_header.r_id = htonl (rid);
-  msg->name_len = htons (name_len);
-  msg->rd_count = htons (rd_count);
-  msg->rd_len = htons (rd_ser_len);
-  msg->reserved = ntohs (0);
+  msg->rd_set_count = htons (rd_set_count);
   msg->private_key = *pkey;
-
-  name_tmp = (char *) &msg[1];
-  GNUNET_memcpy (name_tmp, label, name_len);
-  rd_ser = &name_tmp[name_len];
-  sret = GNUNET_GNSRECORD_records_serialize (rd_count, rd, rd_ser_len, rd_ser);
-  if ((0 > sret) || (sret != rd_ser_len))
+  rd_set = (struct RecordSet*) &msg[1];
+  for (int i = 0; i < rd_set_count; i++)
   {
-    GNUNET_break (0);
-    GNUNET_free (env);
-    return NULL;
+    label = a_label[i];
+    rd = a_rd[i];
+    name_len = strlen (label) + 1;
+    rd_set->name_len = htons (name_len);
+    rd_set->rd_count = htons (a_rd_count[i]);
+    rd_set->rd_len = htons (rd_ser_len[i]);
+    rd_set->reserved = ntohs (0);
+    name_tmp = (char *) &rd_set[1];
+    GNUNET_memcpy (name_tmp, label, name_len);
+    rd_ser = &name_tmp[name_len];
+    sret = GNUNET_GNSRECORD_records_serialize (a_rd_count[i], rd, rd_ser_len[i],
+                                               rd_ser);
+    if ((0 > sret) || (sret != rd_ser_len[i]))
+    {
+      GNUNET_break (0);
+      GNUNET_free (env);
+      return NULL;
+    }
+    // Point to next RecordSet
+    rd_set = (struct RecordSet*) &name_tmp[name_len + rd_ser_len[i]];
   }
-  GNUNET_assert (rd_ser_len == sret);
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Sending NAMESTORE_RECORD_STORE message for name `%s' with %u records\n",
-       label,
-       rd_count);
+       "Sending NAMESTORE_RECORD_STORE message for name %u record sets\n",
+       rd_set_count);
   qe->timeout_task =
     GNUNET_SCHEDULER_add_delayed (NAMESTORE_DELAY_TOLERANCE, &warn_delay, qe);
   if (NULL == h->mq)
@@ -1171,6 +1207,7 @@ GNUNET_NAMESTORE_records_store (
   return qe;
 }
 
+
 static struct GNUNET_NAMESTORE_QueueEntry *
 records_lookup (
   struct GNUNET_NAMESTORE_Handle *h,
diff --git a/src/namestore/perf_namestore_api_postgres.conf b/src/namestore/perf_namestore_api_postgres.conf
index 52d0ecdd5..12fc24d4b 100644
--- a/src/namestore/perf_namestore_api_postgres.conf
+++ b/src/namestore/perf_namestore_api_postgres.conf
@@ -9,3 +9,4 @@ DISABLE = YES
 [namestore-postgres]
 CONFIG = connect_timeout=10 dbname=gnunetcheck
 TEMPORARY_TABLE = YES
+INIT_ON_CONNECT = YES
diff --git a/src/namestore/perf_namestore_api_sqlite.conf b/src/namestore/perf_namestore_api_sqlite.conf
index de0fa3f1f..55c3dc812 100644
--- a/src/namestore/perf_namestore_api_sqlite.conf
+++ b/src/namestore/perf_namestore_api_sqlite.conf
@@ -5,3 +5,4 @@ DISABLE = YES
 
 [namestore-sqlite]
 FILENAME = $GNUNET_TEST_HOME/namestore/sqlite_test.db
+INIT_ON_CONNECT = YES
diff --git a/src/namestore/plugin_namestore_sqlite.c b/src/namestore/plugin_namestore_sqlite.c
index f1097ad0a..35fd340ab 100644
--- a/src/namestore/plugin_namestore_sqlite.c
+++ b/src/namestore/plugin_namestore_sqlite.c
@@ -871,6 +871,7 @@ database_connect (struct Plugin *plugin)
     GNUNET_free (sqlite_filename);
     return GNUNET_SYSERR;
   }
+  GNUNET_free (sqlite_filename);
   GNUNET_break (SQLITE_OK ==
                 sqlite3_busy_timeout (plugin->dbh,
                                       BUSY_TIMEOUT_MS));
diff --git a/src/namestore/plugin_rest_namestore.c b/src/namestore/plugin_rest_namestore.c
index 0475960eb..b9c32e087 100644
--- a/src/namestore/plugin_rest_namestore.c
+++ b/src/namestore/plugin_rest_namestore.c
@@ -721,6 +721,114 @@ ns_lookup_cb (void *cls,
   }
 }
 
+/**
+ * Handle namestore POST import
+ *
+ * @param con_handle the connection handle
+ * @param url the url
+ * @param cls the RequestHandle
+ */
+/*void
+namestore_import (struct GNUNET_REST_RequestHandle *con_handle,
+                         const char *url,
+                         void *cls)
+{
+  struct RequestHandle *handle = cls;
+  struct EgoEntry *ego_entry;
+  struct GNUNET_GNSRECORD_Data *rd_import;
+  char *egoname;
+  json_t *data_js;
+  json_error_t err;
+
+  char term_data[handle->rest_handle->data_size + 1];
+
+  if (0 >= handle->rest_handle->data_size)
+  {
+    handle->response_code = MHD_HTTP_BAD_REQUEST;
+    handle->emsg = GNUNET_strdup (GNUNET_REST_NAMESTORE_NO_DATA);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    return;
+  }
+  term_data[handle->rest_handle->data_size] = '\0';
+  GNUNET_memcpy (term_data,
+                 handle->rest_handle->data,
+                 handle->rest_handle->data_size);
+  data_js = json_loads (term_data, JSON_DECODE_ANY, &err);
+  if (NULL == data_js)
+  {
+    GNUNET_asprintf (&handle->emsg, "Error parsing data: %s", err.text);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    return;
+  }
+  if (!json_is_array (data_js))
+  {
+    handle->emsg = GNUNET_strdup (GNUNET_REST_NAMESTORE_INVALID_DATA);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    json_decref (data_js);
+    return;
+  }
+  size_t num_records = json_array_size (data_js);
+  rd_import = GNUNET_malloc (sizeof (*rd_import) * json_array_size (data_js));
+  for (int i = 0; i < num_records)
+  struct GNUNET_JSON_Specification gnsspec[] =
+  { GNUNET_GNSRECORD_JSON_spec_gnsrecord (&handle->rd, &handle->rd_count,
+                                &handle->record_name),
+    GNUNET_JSON_spec_end () };
+  if (GNUNET_OK != GNUNET_JSON_parse (data_js, gnsspec, NULL, NULL))
+  {
+    handle->emsg = GNUNET_strdup (GNUNET_REST_NAMESTORE_INVALID_DATA);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    json_decref (data_js);
+    return;
+  }
+  GNUNET_JSON_parse_free (gnsspec);
+  if (0 >= strlen (handle->record_name))
+  {
+    handle->response_code = MHD_HTTP_BAD_REQUEST;
+    handle->emsg = GNUNET_strdup (GNUNET_REST_NAMESTORE_INVALID_DATA);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    json_decref (data_js);
+    return;
+  }
+  json_decref (data_js);
+
+  egoname = NULL;
+  ego_entry = NULL;
+
+  // set zone to name if given
+  if (strlen (GNUNET_REST_API_NS_NAMESTORE) + 1 >= strlen (handle->url))
+  {
+    handle->response_code = MHD_HTTP_NOT_FOUND;
+    handle->emsg = GNUNET_strdup (GNUNET_REST_IDENTITY_NOT_FOUND);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    return;
+  }
+  egoname = &handle->url[strlen (GNUNET_REST_API_NS_NAMESTORE) + 1];
+  ego_entry = get_egoentry_namestore (handle, egoname);
+
+  if (NULL == ego_entry)
+  {
+    handle->response_code = MHD_HTTP_NOT_FOUND;
+    handle->emsg = GNUNET_strdup (GNUNET_REST_IDENTITY_NOT_FOUND);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    return;
+  }
+  handle->zone_pkey = GNUNET_IDENTITY_ego_get_private_key (ego_entry->ego);
+  handle->ns_qe = GNUNET_NAMESTORE_records_lookup (ns_handle,
+                                                   handle->zone_pkey,
+                                                   handle->record_name,
+                                                   &ns_lookup_error_cb,
+                                                   handle,
+                                                   &ns_lookup_cb,
+                                                   handle);
+  if (NULL == handle->ns_qe)
+  {
+    handle->emsg = GNUNET_strdup (GNUNET_REST_NAMESTORE_FAILED);
+    GNUNET_SCHEDULER_add_now (&do_error, handle);
+    return;
+  }
+}
+*/
 
 /**
  * Handle namestore POST/PUT request
diff --git a/src/namestore/test_plugin_namestore_postgres.conf b/src/namestore/test_plugin_namestore_postgres.conf
index 2ce8a7792..3b5cb8699 100644
--- a/src/namestore/test_plugin_namestore_postgres.conf
+++ b/src/namestore/test_plugin_namestore_postgres.conf
@@ -1,3 +1,4 @@
 [namestore-postgres]
 CONFIG = connect_timeout=10 dbname=gnunetcheck
+INIT_ON_CONNECT = YES
 TEMPORARY_TABLE = YES
diff --git a/src/namestore/test_plugin_namestore_sqlite.conf b/src/namestore/test_plugin_namestore_sqlite.conf
index 24eecd286..365198db2 100644
--- a/src/namestore/test_plugin_namestore_sqlite.conf
+++ b/src/namestore/test_plugin_namestore_sqlite.conf
@@ -1,2 +1,3 @@
 [namestore-sqlite]
 FILENAME = $GNUNET_TMP/gnunet-test-plugin-namestore-sqlite/sqlite.db
+INIT_ON_CONNECT = YES