1 files changed, 352 insertions, 44 deletions

diff --git a/src/namestore/gnunet-service-namestore.c b/src/namestore/gnunet-service-namestore.c
index 9b2d9b6f3..d735822fb 100644
--- a/src/namestore/gnunet-service-namestore.c
+++ b/src/namestore/gnunet-service-namestore.c
@@ -121,6 +121,23 @@ struct ZoneIteration
   int send_end;
 };
 
+/**
+ * Lock on a record set
+ */
+struct RecordsLock
+{
+  /* DLL */
+  struct RecordsLock *prev;
+
+  /* DLL */
+  struct RecordsLock *next;
+
+  /* Hash of the locked label */
+  struct GNUNET_HashCode label_hash;
+
+  /* Client locking the zone */
+  struct NamestoreClient *client;
+};
 
 /**
  * A namestore client
@@ -394,6 +411,16 @@ static struct StoreActivity *sa_head;
 static struct StoreActivity *sa_tail;
 
 /**
+ * Head of the DLL of record set locks
+ */
+static struct RecordsLock *locks_head;
+
+/**
+ * Tail of the DLL of record set locks
+ */
+static struct RecordsLock *locks_tail;
+
+/**
  * Notification context shared by all monitors.
  */
 static struct GNUNET_NotificationContext *monitor_nc;
@@ -420,6 +447,7 @@ static void
 cleanup_task (void *cls)
 {
   struct CacheOperation *cop;
+  struct RecordsLock *lock;
 
   (void) cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stopping namestore service\n");
@@ -431,6 +459,14 @@ cleanup_task (void *cls)
     GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
     GNUNET_free (cop);
   }
+  while (NULL != (lock = locks_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (locks_head,
+                                 locks_tail,
+                                 lock);
+    GNUNET_free (lock);
+  }
+
   if (NULL != namecache)
   {
     GNUNET_NAMECACHE_disconnect (namecache);
@@ -808,7 +844,8 @@ send_lookup_response (struct NamestoreClient *nc,
  * @param rid client's request ID
  */
 static void
-send_store_response (struct NamestoreClient *nc, int res, uint32_t rid)
+send_store_response (struct NamestoreClient *nc, int res, const char*emsg,
+                     uint32_t rid)
 {
   struct GNUNET_MQ_Envelope *env;
   struct RecordStoreResponseMessage *rcr_msg;
@@ -820,10 +857,17 @@ send_store_response (struct NamestoreClient *nc, int res, uint32_t rid)
                             "Store requests completed",
                             1,
                             GNUNET_NO);
-  env = GNUNET_MQ_msg (rcr_msg,
-                       GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE_RESPONSE);
+  env = GNUNET_MQ_msg_extra (rcr_msg,
+                             (NULL != emsg) ? strlen (emsg) + 1 : 0,
+                             GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE_RESPONSE);
   rcr_msg->gns_header.r_id = htonl (rid);
   rcr_msg->op_result = htonl (res);
+  rcr_msg->reserved = htons (0);
+  if (NULL != emsg)
+  {
+    rcr_msg->emsg_len = htons (strlen (emsg) + 1);
+    memcpy (&rcr_msg[1], emsg, strlen (emsg) + 1);
+  }
   GNUNET_MQ_send (nc->mq, env);
 }
 
@@ -874,7 +918,7 @@ finish_cache_operation (void *cls, int32_t success, const char *emsg)
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CACHE operation completed\n");
   GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
   if (NULL != cop->nc)
-    send_store_response (cop->nc, success, cop->rid);
+    send_store_response (cop->nc, success, emsg, cop->rid);
   if (NULL != (zi = cop->zi))
   {
     zi->cache_ops--;
@@ -910,29 +954,40 @@ refresh_block (struct NamestoreClient *nc,
                const struct GNUNET_GNSRECORD_Data *rd)
 {
   struct GNUNET_GNSRECORD_Block *block;
+  struct GNUNET_GNSRECORD_Data rd_clean[rd_count];
   struct CacheOperation *cop;
   struct GNUNET_IDENTITY_PublicKey pkey;
   struct GNUNET_GNSRECORD_Data *nick;
   struct GNUNET_GNSRECORD_Data *res;
   unsigned int res_count;
+  unsigned int rd_count_clean;
   struct GNUNET_TIME_Absolute exp_time;
 
+  /** Do not block-cache tombstones */
+  rd_count_clean = 0;
+  for (int i = 0; i < rd_count; i++)
+  {
+    if (GNUNET_GNSRECORD_TYPE_TOMBSTONE == rd[i].record_type)
+      continue;
+    rd_clean[rd_count_clean++] = rd[i];
+  }
+
   nick = get_nick_record (zone_key);
-  res_count = rd_count;
-  res = (struct GNUNET_GNSRECORD_Data *) rd;  /* fixme: a bit unclean... */
+  res_count = rd_count_clean;
+  res = (struct GNUNET_GNSRECORD_Data *) rd_clean;  /* fixme: a bit unclean... */
   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_count, rd, &res_count, &res);
+    merge_with_nick_records (nick, rd_count_clean, rd_clean, &res_count, &res);
   }
   if (NULL != nick)
     GNUNET_free (nick);
   if (0 == res_count)
   {
     if (NULL != nc)
-      send_store_response (nc, GNUNET_OK, rid);
-    if (rd != res)
+      send_store_response (nc, GNUNET_OK, NULL, rid);
+    if (rd_clean != res)
       GNUNET_free (res);
     return;   /* no data, no need to update cache */
   }
@@ -943,20 +998,21 @@ refresh_block (struct NamestoreClient *nc,
                               1,
                               GNUNET_NO);
     if (NULL != nc)
-      send_store_response (nc, GNUNET_OK, rid);
-    if (rd != res)
+      send_store_response (nc, GNUNET_OK, NULL, rid);
+    if (rd_clean != res)
       GNUNET_free (res);
     return;
   }
-  exp_time = GNUNET_GNSRECORD_record_get_expiration_time (res_count, res);
+  exp_time = GNUNET_GNSRECORD_record_get_expiration_time (res_count, res,
+                                                          GNUNET_TIME_UNIT_ZERO_ABS);
   if (cache_keys)
     GNUNET_assert (GNUNET_OK ==
-      GNUNET_GNSRECORD_block_create2 (zone_key, exp_time, name,
-                                      res, res_count, &block));
+                   GNUNET_GNSRECORD_block_create2 (zone_key, exp_time, name,
+                                                   res, res_count, &block));
   else
     GNUNET_assert (GNUNET_OK ==
-      GNUNET_GNSRECORD_block_create (zone_key, exp_time, name,
-                                     res, res_count, &block));
+                   GNUNET_GNSRECORD_block_create (zone_key, exp_time, name,
+                                                  res, res_count, &block));
   GNUNET_assert (NULL != block);
   GNUNET_IDENTITY_key_get_public (zone_key, &pkey);
   GNUNET_log (
@@ -974,7 +1030,7 @@ refresh_block (struct NamestoreClient *nc,
   cop->nc = nc;
   cop->zi = zi;
   if (NULL != zi)
-    zi->cache_ops++;
+    zi->cache_ops ++;
   cop->rid = rid;
   GNUNET_CONTAINER_DLL_insert (cop_head, cop_tail, cop);
   cop->qe = GNUNET_NAMECACHE_block_cache (namecache,
@@ -982,7 +1038,7 @@ refresh_block (struct NamestoreClient *nc,
                                           &finish_cache_operation,
                                           cop);
   GNUNET_free (block);
-  if (rd != res)
+  if (rd_clean != res)
     GNUNET_free (res);
 }
 
@@ -1098,6 +1154,7 @@ client_disconnect_cb (void *cls,
   struct NamestoreClient *nc = app_ctx;
   struct ZoneIteration *no;
   struct CacheOperation *cop;
+  struct RecordsLock *lock;
 
   (void) cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p disconnected\n", client);
@@ -1148,6 +1205,15 @@ client_disconnect_cb (void *cls,
   for (cop = cop_head; NULL != cop; cop = cop->next)
     if (nc == cop->nc)
       cop->nc = NULL;
+  for (lock = locks_head; NULL != lock; lock = lock->next)
+  {
+    if (nc != lock->client)
+      continue;
+    GNUNET_CONTAINER_DLL_remove (locks_head,
+                                 locks_tail,
+                                 lock);
+    GNUNET_free (lock);
+  }
   GNUNET_free (nc);
 }
 
@@ -1341,6 +1407,105 @@ check_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   return GNUNET_OK;
 }
 
+static void
+calculate_lock_hash (const char *label,
+                     const struct GNUNET_IDENTITY_PrivateKey *zone,
+                     struct GNUNET_HashCode *result)
+{
+  struct GNUNET_HashContext *hctx;
+
+  hctx = GNUNET_CRYPTO_hash_context_start ();
+  GNUNET_CRYPTO_hash_context_read (hctx, label, strlen (label));
+  GNUNET_CRYPTO_hash_context_read (hctx, zone,
+                                   sizeof (*zone));
+  GNUNET_CRYPTO_hash_context_finish (hctx, result);
+}
+
+/**
+ * Release a lock on a record set.
+ * Does nothing if lock not held.
+ *
+ * @param label the label of the record set
+ * @param zone the zone
+ * @param nc the client releasing the lock
+ */
+static void
+NST_label_lock_release (const char *label,
+                        const struct GNUNET_IDENTITY_PrivateKey *zone,
+                        const struct NamestoreClient *nc)
+{
+  struct GNUNET_HashCode label_hash;
+  struct RecordsLock *lock;
+
+  calculate_lock_hash (label, zone, &label_hash);
+  for (lock = locks_head; NULL != lock; lock = lock->next)
+    if (0 == memcmp (&label_hash, &lock->label_hash, sizeof (label_hash)))
+      break;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Record locked: %s\n", (NULL == lock) ? "No" : "Yes");
+  if (NULL == lock)
+    return;
+  if (lock->client != nc)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "Lock is held by other client on `%s'\n", label);
+    return;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Unocking %s\n", GNUNET_h2s (&label_hash));
+  GNUNET_CONTAINER_DLL_remove (locks_head,
+                               locks_tail,
+                               lock);
+  GNUNET_free (lock);
+}
+
+/**
+ * Get/set a lock on a record set.
+ * May be called multiple times but will
+ * not aquire additional locks.
+ *
+ * @param the label of the record set
+ * @param the zone
+ * @param the client doing the locking
+ * @return GNUNET_YES if lock retrieved or set already.
+ */
+static enum GNUNET_GenericReturnValue
+NST_label_lock (const char *label,
+                const struct GNUNET_IDENTITY_PrivateKey *zone,
+                struct NamestoreClient *nc)
+{
+  struct GNUNET_HashCode label_hash;
+  struct RecordsLock *lock;
+
+  calculate_lock_hash (label, zone, &label_hash);
+  for (lock = locks_head; NULL != lock; lock = lock->next)
+    if (0 == memcmp (&label_hash, &lock->label_hash, sizeof (label_hash)))
+      break;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Record locked: %s\n", (NULL == lock) ? "No" : "Yes");
+  if (NULL != lock)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Client holds lock: %s\n", (lock->client != nc) ? "No" : "Yes");
+    if (lock->client != nc)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Lock is held by other client on `%s'\n", label);
+      return GNUNET_NO;
+    }
+    return GNUNET_YES;
+  }
+  lock = GNUNET_new (struct RecordsLock);
+  lock->client = nc;
+  memcpy (&lock->label_hash, &label_hash, sizeof (label_hash));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Locking %s\n", GNUNET_h2s (&label_hash));
+  GNUNET_CONTAINER_DLL_insert (locks_head,
+                               locks_tail,
+                               lock);
+  return GNUNET_YES;
+}
+
 
 /**
  * Handles a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP message
@@ -1355,13 +1520,14 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   struct GNUNET_MQ_Envelope *env;
   struct LabelLookupResponseMessage *llr_msg;
   struct RecordLookupContext rlc;
+  struct RecordsLock *lock;
+  struct GNUNET_HashCode label_hash;
   const char *name_tmp;
   char *res_name;
   char *conv_name;
   uint32_t name_len;
   int res;
 
-  name_len = ntohl (ll_msg->label_len);
   name_tmp = (const char *) &ll_msg[1];
   GNUNET_SERVICE_client_continue (nc->client);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1377,6 +1543,29 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
     GNUNET_SERVICE_client_drop (nc->client);
     return;
   }
+  name_len = strlen (conv_name) + 1;
+  if (GNUNET_YES == ntohl (ll_msg->locking))
+  {
+    if (GNUNET_NO == NST_label_lock (conv_name, &ll_msg->zone, nc))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Lock is held by other client on `%s'\n", conv_name);
+      env =
+        GNUNET_MQ_msg_extra (llr_msg,
+                             name_len,
+                             GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_LOOKUP_RESPONSE);
+      llr_msg->gns_header.r_id = ll_msg->gns_header.r_id;
+      llr_msg->private_key = ll_msg->zone;
+      llr_msg->name_len = htons (name_len);
+      llr_msg->rd_count = htons (0);
+      llr_msg->rd_len = htons (0);
+      llr_msg->found = htons (GNUNET_SYSERR);
+      GNUNET_memcpy (&llr_msg[1], conv_name, name_len);
+      GNUNET_MQ_send (nc->mq, env);
+      GNUNET_free (conv_name);
+      return;
+    }
+  }
   rlc.label = conv_name;
   rlc.found = GNUNET_NO;
   rlc.res_rd_count = 0;
@@ -1388,7 +1577,6 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
                                       conv_name,
                                       &lookup_it,
                                       &rlc);
-  GNUNET_free (conv_name);
   env =
     GNUNET_MQ_msg_extra (llr_msg,
                          name_len + rlc.rd_ser_len,
@@ -1400,16 +1588,18 @@ handle_record_lookup (void *cls, const struct LabelLookupMessage *ll_msg)
   llr_msg->rd_len = htons (rlc.rd_ser_len);
   res_name = (char *) &llr_msg[1];
   if ((GNUNET_YES == rlc.found) && (GNUNET_OK == res))
-    llr_msg->found = ntohs (GNUNET_YES);
+    llr_msg->found = htons (GNUNET_YES);
   else
-    llr_msg->found = ntohs (GNUNET_NO);
-  GNUNET_memcpy (&llr_msg[1], name_tmp, name_len);
+    llr_msg->found = htons (GNUNET_NO);
+  GNUNET_memcpy (&llr_msg[1], 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.res_rd);
+  GNUNET_free (conv_name);
 }
 
 
+
 /**
  * Checks a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE message
  *
@@ -1452,6 +1642,45 @@ check_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
 
 
 /**
+ * Check if set contains a tombstone, store if necessary
+ *
+ * @param cls a `struct GNUNET_GNSRECORD_Data **` for storing the nick (if found)
+ * @param seq sequence number of the record, MUST NOT BE ZERO
+ * @param private_key the private key of the zone (unused)
+ * @param label should be #GNUNET_GNS_EMPTY_LABEL_AT
+ * @param rd_count number of records in @a rd
+ * @param rd records stored under @a label in the zone
+ */
+static void
+get_block_exp_existing (void *cls,
+                        uint64_t seq,
+                        const struct
+                        GNUNET_IDENTITY_PrivateKey *private_key,
+                        const char *label,
+                        unsigned int rd_count,
+                        const struct GNUNET_GNSRECORD_Data *rd)
+{
+  struct GNUNET_TIME_Absolute *exp = cls;
+  struct GNUNET_GNSRECORD_Data rd_pub[rd_count];
+  unsigned int rd_pub_count;
+  char *emsg;
+
+  if (GNUNET_OK != GNUNET_GNSRECORD_convert_records_for_export (label,
+                                                                rd,
+                                                                rd_count,
+                                                                rd_pub,
+                                                                &rd_pub_count,
+                                                                exp,
+                                                                &emsg))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%s\n", emsg);
+    GNUNET_free (emsg);
+  }
+}
+
+
+/**
  * Handles a #GNUNET_MESSAGE_TYPE_NAMESTORE_RECORD_STORE message
  *
  * @param cls client sending the message
@@ -1470,37 +1699,75 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
   unsigned int rd_count;
   int res;
   struct StoreActivity *sa;
+  struct RecordsLock *lock;
+  struct GNUNET_HashCode label_hash;
+  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);
-  GNUNET_break (0 == ntohs (rp_msg->reserved));
   name_tmp = (const char *) &rp_msg[1];
   rd_ser = &name_tmp[name_len];
   {
     struct GNUNET_GNSRECORD_Data rd[GNUNET_NZL (rd_count)];
-
-    if (GNUNET_OK !=
-        GNUNET_GNSRECORD_records_deserialize (rd_ser_len, rd_ser, rd_count, rd))
-    {
-      GNUNET_break (0);
-      GNUNET_SERVICE_client_drop (nc->client);
-      return;
-    }
+    char *emsg;
 
     /* Extracting and converting private key */
     conv_name = GNUNET_GNSRECORD_string_normalize (name_tmp);
     if (NULL == conv_name)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Error converting name `%s'\n",
+                  "Error normalizing name `%s'\n",
                   name_tmp);
-      GNUNET_SERVICE_client_drop (nc->client);
+      send_store_response (nc, GNUNET_SYSERR, _ ("Error normalizing name."),
+                           rid);
+      GNUNET_SERVICE_client_continue (nc->client);
+      return;
+    }
+
+    /* 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);
+      send_store_response (nc, GNUNET_SYSERR, emsg, rid);
+      GNUNET_free (emsg);
+      GNUNET_free (conv_name);
+      GNUNET_SERVICE_client_continue (nc->client);
+      return;
+    }
+
+    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;
     }
+    if (GNUNET_YES == ntohl (rp_msg->locking))
+    {
+      if (GNUNET_NO == NST_label_lock (conv_name, &rp_msg->private_key, nc))
+      {
+        send_store_response (nc, GNUNET_SYSERR, _ ("Record set locked."), rid);
+        GNUNET_SERVICE_client_continue (nc->client);
+        GNUNET_free (conv_name);
+        return;
+      }
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Client has lock on `%s', continuing.\n", conv_name);
+      if (GNUNET_YES == ntohl (rp_msg->locking))
+        NST_label_lock_release (conv_name, &rp_msg->private_key, nc);
+    }
+
     GNUNET_STATISTICS_update (statistics,
                               "Well-formed store requests received",
                               1,
@@ -1509,12 +1776,12 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
                 "Creating %u records for name `%s'\n",
                 (unsigned int) rd_count,
                 conv_name);
-    if ((0 == rd_count) &&
-        (GNUNET_NO == GSN_database->lookup_records (GSN_database->cls,
+    if ((GNUNET_NO == GSN_database->lookup_records (GSN_database->cls,
                                                     &rp_msg->private_key,
                                                     conv_name,
-                                                    NULL,
-                                                    0)))
+                                                    &get_block_exp_existing,
+                                                    &existing_block_exp)) &&
+        (rd_count == 0))
     {
       /* This name does not exist, so cannot be removed */
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1525,9 +1792,18 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
     else
     {
       /* remove "NICK" records, unless this is for the
-       #GNUNET_GNS_EMPTY_LABEL_AT label */
+       #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;
+      char *emsg;
       int have_nick;
 
       rd_clean_off = 0;
@@ -1535,6 +1811,7 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       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++;
@@ -1546,6 +1823,38 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
           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_YES,
+                                                              &emsg))
+      {
+        send_store_response (nc, GNUNET_SYSERR, emsg, rid);
+        GNUNET_free (emsg);
+        GNUNET_SERVICE_client_continue (nc->client);
+        GNUNET_free (conv_name);
+        return;
+      }
+      /*
+       * If existing_block_exp is 0, then there was not 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, <=, existing_block_exp))
+      {
+        rd_nf[rd_nf_count].record_type = GNUNET_GNSRECORD_TYPE_TOMBSTONE;
+        rd_nf[rd_nf_count].expiration_time =
+          existing_block_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;
+        rd_nf_count++;
+      }
       if ((0 == strcmp (GNUNET_GNS_EMPTY_LABEL_AT, conv_name)) &&
           (GNUNET_NO == have_nick))
       {
@@ -1555,19 +1864,18 @@ handle_record_store (void *cls, const struct RecordStoreMessage *rp_msg)
       res = GSN_database->store_records (GSN_database->cls,
                                          &rp_msg->private_key,
                                          conv_name,
-                                         rd_clean_off,
-                                         rd_clean);
+                                         rd_nf_count,
+                                         rd_nf);
     }
 
     if (GNUNET_OK != res)
     {
-      /* store not successful, not need to tell monitors */
-      send_store_response (nc, res, rid);
+      /* store not successful, no need to tell monitors */
+      send_store_response (nc, res, _ ("Store failed"), rid);
       GNUNET_SERVICE_client_continue (nc->client);
       GNUNET_free (conv_name);
       return;
     }
-
     sa = GNUNET_malloc (sizeof(struct StoreActivity)
                         + ntohs (rp_msg->gns_header.header.size));
     GNUNET_CONTAINER_DLL_insert (sa_head, sa_tail, sa);