-ensure stats queues do not grow too big
[oweals/gnunet.git] / src / multicast / gnunet-service-multicast.c
index f4e37865fb5697d137fb022185b4c5010ce30415..94f9d2f88e8ce276430f830747feffae267cc7b8 100644 (file)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     (C) 2009 Christian Grothoff (and other contributing authors)
+     Copyright (C) 2009 GNUnet e.V.
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -14,8 +14,8 @@
 
      You should have received a copy of the GNU General Public License
      along with GNUnet; see the file COPYING.  If not, write to the
-     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
-     Boston, MA 02111-1307, USA.
+     Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+     Boston, MA 02110-1301, USA.
 */
 
 /**
@@ -88,17 +88,32 @@ static struct GNUNET_CONTAINER_MultiHashMap *members;
 static struct GNUNET_CONTAINER_MultiHashMap *group_members;
 
 /**
- * Incoming CADET channels.
+ * Incoming CADET channels with connected children in the tree.
  * Group's pub_key_hash -> struct Channel * (multi)
  */
 static struct GNUNET_CONTAINER_MultiHashMap *channels_in;
 
 /**
- * Outgoing CADET channels.
+ * Outgoing CADET channels connecting to parents in the tree.
  * Group's pub_key_hash -> struct Channel * (multi)
  */
 static struct GNUNET_CONTAINER_MultiHashMap *channels_out;
 
+/**
+ * Incoming replay requests from CADET.
+ * Group's pub_key_hash ->
+ *   H(fragment_id, message_id, fragment_offset, flags) -> struct Channel *
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *replay_req_cadet;
+
+/**
+ * Incoming replay requests from clients.
+ * Group's pub_key_hash ->
+ *   H(fragment_id, message_id, fragment_offset, flags) -> struct GNUNET_SERVER_Client *
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *replay_req_client;
+
+
 /**
  * Join status of a remote peer.
  */
@@ -142,17 +157,17 @@ struct Channel
   /**
    * Public key of the target group.
    */
-  struct GNUNET_CRYPTO_EddsaPublicKey group_key;
+  struct GNUNET_CRYPTO_EddsaPublicKey group_pub_key;
 
   /**
-   * Hash of @a group_key.
+   * Hash of @a group_pub_key.
    */
-  struct GNUNET_HashCode group_key_hash;
+  struct GNUNET_HashCode group_pub_hash;
 
   /**
    * Public key of the joining member.
    */
-  struct GNUNET_CRYPTO_EcdsaPublicKey member_key;
+  struct GNUNET_CRYPTO_EcdsaPublicKey member_pub_key;
 
   /**
    * Remote peer identity.
@@ -165,6 +180,11 @@ struct Channel
    */
   int8_t join_status;
 
+  /**
+   * Number of messages waiting to be sent to CADET.
+   */
+  uint8_t msgs_pending;
+
   /**
    * Channel direction.
    * @see enum ChannelDirection
@@ -294,14 +314,22 @@ struct Member
 };
 
 
+struct ReplayRequestKey
+{
+  uint64_t fragment_id;
+  uint64_t message_id;
+  uint64_t fragment_offset;
+  uint64_t flags;
+};
+
+
 /**
  * Task run during shutdown.
  *
  * @param cls unused
- * @param tc unused
  */
 static void
-shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+shutdown_task (void *cls)
 {
   if (NULL != core)
   {
@@ -375,6 +403,97 @@ cleanup_group (struct Group *grp)
 }
 
 
+void
+replay_key_hash (uint64_t fragment_id, uint64_t message_id,
+                 uint64_t fragment_offset, uint64_t flags,
+                 struct GNUNET_HashCode *key_hash)
+{
+  struct ReplayRequestKey key = {
+    .fragment_id = fragment_id,
+    .message_id = message_id,
+    .fragment_offset = fragment_offset,
+    .flags = flags,
+  };
+  GNUNET_CRYPTO_hash (&key, sizeof (key), key_hash);
+}
+
+
+/**
+ * Remove channel from replay request hashmap.
+ *
+ * @param chn
+ *        Channel to remove.
+ *
+ * @return #GNUNET_YES if there are more entries to process,
+ *         #GNUNET_NO when reached end of hashmap.
+ */
+static int
+replay_req_remove_cadet (struct Channel *chn)
+{
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
+                                                        &chn->grp->pub_key_hash);
+  if (NULL == grp_replay_req)
+    return GNUNET_NO;
+
+  struct GNUNET_CONTAINER_MultiHashMapIterator *
+    it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
+  struct GNUNET_HashCode key;
+  const struct Channel *c;
+  while (GNUNET_YES
+         == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
+                                                         (const void **) &c))
+  {
+    if (c == chn)
+    {
+      GNUNET_CONTAINER_multihashmap_remove (grp_replay_req, &key, chn);
+      GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
+      return GNUNET_YES;
+    }
+  }
+  GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
+  return GNUNET_NO;
+}
+
+
+/**
+ * Remove client from replay request hashmap.
+ *
+ * @param client
+ *        Client to remove.
+ *
+ * @return #GNUNET_YES if there are more entries to process,
+ *         #GNUNET_NO when reached end of hashmap.
+ */
+static int
+replay_req_remove_client (struct Group *grp, struct GNUNET_SERVER_Client *client)
+{
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
+                                                        &grp->pub_key_hash);
+  if (NULL == grp_replay_req)
+    return GNUNET_NO;
+
+  struct GNUNET_CONTAINER_MultiHashMapIterator *
+    it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
+  struct GNUNET_HashCode key;
+  const struct GNUNET_SERVER_Client *c;
+  while (GNUNET_YES
+         == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
+                                                         (const void **) &c))
+  {
+    if (c == client)
+    {
+      GNUNET_CONTAINER_multihashmap_remove (replay_req_client, &key, client);
+      GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
+      return GNUNET_YES;
+    }
+  }
+  GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
+  return GNUNET_NO;
+}
+
+
 /**
  * Called whenever a client is disconnected.
  *
@@ -396,7 +515,7 @@ client_notify_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "%p User context is NULL in client_disconnect()\n", grp);
-    GNUNET_assert (0);
+    GNUNET_break (0);
     return;
   }
 
@@ -417,6 +536,8 @@ client_notify_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
     cl = cl->next;
   }
 
+  while (GNUNET_YES == replay_req_remove_client (grp, client));
+
   if (NULL == grp->clients_head)
   { /* Last client disconnected. */
 #if FIXME
@@ -433,15 +554,30 @@ client_notify_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 }
 
 
+/**
+ * Send message to a client.
+ */
+static void
+client_send (struct GNUNET_SERVER_Client *client,
+             const struct GNUNET_MessageHeader *msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "%p Sending message to client.\n", client);
+
+  GNUNET_SERVER_notification_context_add (nc, client);
+  GNUNET_SERVER_notification_context_unicast (nc, client, msg, GNUNET_NO);
+}
+
+
 /**
  * Send message to all clients connected to the group.
  */
 static void
-client_send_msg (const struct Group *grp,
-                 const struct GNUNET_MessageHeader *msg)
+client_send_group (const struct Group *grp,
+                   const struct GNUNET_MessageHeader *msg)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "%p Sending message to clients.\n", grp);
+              "%p Sending message to all clients of the group.\n", grp);
 
   struct ClientList *cl = grp->clients_head;
   while (NULL != cl)
@@ -463,7 +599,7 @@ client_send_origin_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
   const struct GNUNET_MessageHeader *msg = cls;
   struct Member *orig = origin;
 
-  client_send_msg (&orig->grp, msg);
+  client_send_group (&orig->grp, msg);
   return GNUNET_YES;
 }
 
@@ -480,7 +616,7 @@ client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 
   if (NULL != mem->join_dcsn)
   { /* Only send message to admitted members */
-    client_send_msg (&mem->grp, msg);
+    client_send_group (&mem->grp, msg);
   }
   return GNUNET_YES;
 }
@@ -489,45 +625,93 @@ client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 /**
  * Send message to all origin and member clients connected to the group.
  *
- * @param grp  The group to send @a msg to.
- * @param msg  Message to send.
+ * @param pub_key_hash
+ *        H(key_pub) of the group.
+ * @param msg
+ *        Message to send.
  */
 static int
 client_send_all (struct GNUNET_HashCode *pub_key_hash,
                  const struct GNUNET_MessageHeader *msg)
 {
   int n = 0;
-  if (origins != NULL)
-    n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
-                                                     client_send_origin_cb,
-                                                     (void *) msg);
-  if (members != NULL)
-    n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash,
-                                                     client_send_member_cb,
-                                                     (void *) msg);
+  n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
+                                                   client_send_origin_cb,
+                                                   (void *) msg);
+  n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash,
+                                                   client_send_member_cb,
+                                                   (void *) msg);
   return n;
 }
 
 
 /**
- * Send message to all origin clients connected to the group.
+ * Send message to a random origin client or a random member client.
  *
  * @param grp  The group to send @a msg to.
  * @param msg  Message to send.
  */
 static int
+client_send_random (struct GNUNET_HashCode *pub_key_hash,
+                    const struct GNUNET_MessageHeader *msg)
+{
+  int n = 0;
+  n = GNUNET_CONTAINER_multihashmap_get_random (origins, client_send_origin_cb,
+                                                 (void *) msg);
+  if (n <= 0)
+    n = GNUNET_CONTAINER_multihashmap_get_random (members, client_send_member_cb,
+                                                   (void *) msg);
+  return n;
+}
+
+
+/**
+ * Send message to all origin clients connected to the group.
+ *
+ * @param pub_key_hash
+ *        H(key_pub) of the group.
+ * @param msg
+ *        Message to send.
+ */
+static int
 client_send_origin (struct GNUNET_HashCode *pub_key_hash,
                     const struct GNUNET_MessageHeader *msg)
 {
   int n = 0;
-  if (origins != NULL)
-    n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
-                                                     client_send_origin_cb,
-                                                     (void *) msg);
+  n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
+                                                   client_send_origin_cb,
+                                                   (void *) msg);
   return n;
 }
 
 
+/**
+ * Send fragment acknowledgement to all clients of the channel.
+ *
+ * @param pub_key_hash
+ *        H(key_pub) of the group.
+ */
+static void
+client_send_ack (struct GNUNET_HashCode *pub_key_hash)
+{
+  static struct GNUNET_MessageHeader *msg = NULL;
+  if (NULL == msg)
+  {
+    msg = GNUNET_malloc (sizeof (*msg));
+    msg->type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK);
+    msg->size = htons (sizeof (*msg));
+  }
+  client_send_all (pub_key_hash, msg);
+}
+
+
+struct CadetTransmitClosure
+{
+  struct Channel *chn;
+  const struct GNUNET_MessageHeader *msg;
+};
+
+
 /**
  * CADET is ready to transmit a message.
  */
@@ -539,10 +723,21 @@ cadet_notify_transmit_ready (void *cls, size_t buf_size, void *buf)
     /* FIXME: connection closed */
     return 0;
   }
-  const struct GNUNET_MessageHeader *msg = cls;
-  uint16_t msg_size = ntohs (msg->size);
+  struct CadetTransmitClosure *tcls = cls;
+  struct Channel *chn = tcls->chn;
+  uint16_t msg_size = ntohs (tcls->msg->size);
   GNUNET_assert (msg_size <= buf_size);
-  memcpy (buf, msg, msg_size);
+  memcpy (buf, tcls->msg, msg_size);
+  GNUNET_free (tcls);
+
+  if (0 == chn->msgs_pending)
+  {
+    GNUNET_break (0);
+  }
+  else if (0 == --chn->msgs_pending)
+  {
+    client_send_ack (&chn->group_pub_hash);
+  }
   return msg_size;
 }
 
@@ -554,14 +749,19 @@ cadet_notify_transmit_ready (void *cls, size_t buf_size, void *buf)
  * @param msg  Message.
  */
 static void
-cadet_send_msg (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
+cadet_send_channel (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
 {
+  struct CadetTransmitClosure *tcls = GNUNET_malloc (sizeof (*tcls));
+  tcls->chn = chn;
+  tcls->msg = msg;
+
+  chn->msgs_pending++;
   chn->tmit_handle
     = GNUNET_CADET_notify_transmit_ready (chn->channel, GNUNET_NO,
                                           GNUNET_TIME_UNIT_FOREVER_REL,
                                           ntohs (msg->size),
                                           &cadet_notify_transmit_ready,
-                                          (void *) msg);
+                                          tcls);
   GNUNET_assert (NULL != chn->tmit_handle);
 }
 
@@ -571,10 +771,10 @@ cadet_send_msg (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
  *
  * @param peer
  *        Peer to connect to.
- * @param group_key
+ * @param group_pub_key
  *        Public key of group the channel belongs to.
- * @param group_key_hash
- *        Hash of @a group_key.
+ * @param group_pub_hash
+ *        Hash of @a group_pub_key.
  *
  * @return Channel.
  */
@@ -583,15 +783,15 @@ cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer)
 {
   struct Channel *chn = GNUNET_malloc (sizeof (*chn));
   chn->grp = grp;
-  chn->group_key = grp->pub_key;
-  chn->group_key_hash = grp->pub_key_hash;
+  chn->group_pub_key = grp->pub_key;
+  chn->group_pub_hash = grp->pub_key_hash;
   chn->peer = *peer;
   chn->direction = DIR_OUTGOING;
   chn->join_status = JOIN_WAITING;
   chn->channel = GNUNET_CADET_channel_create (cadet, chn, &chn->peer,
                                               GNUNET_APPLICATION_TYPE_MULTICAST,
                                               GNUNET_CADET_OPTION_RELIABLE);
-  GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_key_hash, chn,
+  GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_pub_hash, chn,
                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
   return chn;
 }
@@ -604,30 +804,30 @@ static void
 cadet_send_join_request (struct Member *mem)
 {
   mem->origin_channel = cadet_channel_create (&mem->grp, &mem->origin);
-  cadet_send_msg (mem->origin_channel, &mem->join_req->header);
+  cadet_send_channel (mem->origin_channel, &mem->join_req->header);
 
   uint32_t i;
   for (i = 0; i < mem->relay_count; i++)
   {
     struct Channel *
       chn = cadet_channel_create (&mem->grp, &mem->relays[i]);
-    cadet_send_msg (chn, &mem->join_req->header);
+    cadet_send_channel (chn, &mem->join_req->header);
   }
 }
 
 
 static int
 cadet_send_join_decision_cb (void *cls,
-                             const struct GNUNET_HashCode *group_key_hash,
+                             const struct GNUNET_HashCode *group_pub_hash,
                              void *channel)
 {
   const struct MulticastJoinDecisionMessageHeader *hdcsn = cls;
   struct Channel *chn = channel;
 
-  if (0 == memcmp (&hdcsn->member_key, &chn->member_key, sizeof (chn->member_key))
+  if (0 == memcmp (&hdcsn->member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key))
       && 0 == memcmp (&hdcsn->peer, &chn->peer, sizeof (chn->peer)))
   {
-    cadet_send_msg (chn, &hdcsn->header);
+    cadet_send_channel (chn, &hdcsn->header);
     return GNUNET_NO;
   }
   return GNUNET_YES;
@@ -651,29 +851,47 @@ cadet_send_join_decision (struct Group *grp,
  * Iterator callback for sending a message to origin clients.
  */
 static int
-cadet_send_members_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
-                       void *channel)
+cadet_send_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
+               void *channel)
 {
   const struct GNUNET_MessageHeader *msg = cls;
   struct Channel *chn = channel;
   if (JOIN_ADMITTED == chn->join_status)
-    cadet_send_msg (chn, msg);
+    cadet_send_channel (chn, msg);
   return GNUNET_YES;
 }
 
 
+/**
+ * Send message to all connected children.
+ */
 static int
-cadet_send_members (struct GNUNET_HashCode *pub_key_hash,
-                    const struct GNUNET_MessageHeader *msg)
+cadet_send_children (struct GNUNET_HashCode *pub_key_hash,
+                     const struct GNUNET_MessageHeader *msg)
 {
   int n = 0;
   if (channels_in != NULL)
     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, pub_key_hash,
-                                                     cadet_send_members_cb,
-                                                     (void *) msg);
+                                                     cadet_send_cb, (void *) msg);
   return n;
 }
 
+
+/**
+ * Send message to all connected parents.
+ */
+static int
+cadet_send_parents (struct GNUNET_HashCode *pub_key_hash,
+                    const struct GNUNET_MessageHeader *msg)
+{
+  int n = 0;
+  if (channels_in != NULL)
+    n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_out, pub_key_hash,
+                                                     cadet_send_cb, (void *) msg);
+  return n;
+}
+
+
 /**
  * Handle a connecting client starting an origin.
  */
@@ -741,7 +959,7 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
 
   GNUNET_CRYPTO_ecdsa_key_get_public (&msg->member_key, &mem_pub_key);
   GNUNET_CRYPTO_hash (&mem_pub_key, sizeof (mem_pub_key), &mem_pub_key_hash);
-  GNUNET_CRYPTO_hash (&msg->group_key, sizeof (msg->group_key), &pub_key_hash);
+  GNUNET_CRYPTO_hash (&msg->group_pub_key, sizeof (msg->group_pub_key), &pub_key_hash);
 
   struct GNUNET_CONTAINER_MultiHashMap *
     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members, &pub_key_hash);
@@ -762,7 +980,7 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
 
     grp = &mem->grp;
     grp->is_origin = GNUNET_NO;
-    grp->pub_key = msg->group_key;
+    grp->pub_key = msg->group_pub_key;
     grp->pub_key_hash = pub_key_hash;
 
     if (NULL == grp_mem)
@@ -788,9 +1006,11 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "%p Client connected to group %s..\n",
               mem, GNUNET_h2s (&grp->pub_key_hash));
+  char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "%p ..as member %s.\n",
-              mem, GNUNET_h2s (&mem_pub_key_hash));
+              "%p ..as member %s (%s).\n",
+              mem, GNUNET_h2s (&mem->pub_key_hash), str);
+  GNUNET_free (str);
 
   GNUNET_SERVER_client_set_user_context (client, grp);
 
@@ -802,10 +1022,10 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
                                                 mem->join_dcsn,
                                                 GNUNET_NO);
   }
-  else if (grp->clients_head == grp->clients_tail)
+  else
   { /* First client of the group, send join request. */
     struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
-    uint32_t relay_count = ntohs (msg->relay_count);
+    uint32_t relay_count = ntohl (msg->relay_count);
     uint16_t relay_size = relay_count * sizeof (*relays);
     struct GNUNET_MessageHeader *join_msg = NULL;
     uint16_t join_msg_size = 0;
@@ -827,12 +1047,13 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
       req = GNUNET_malloc (sizeof (*req) + join_msg_size);
     req->header.size = htons (sizeof (*req) + join_msg_size);
     req->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST);
-    req->group_key = grp->pub_key;
+    req->group_pub_key = grp->pub_key;
     req->peer = this_peer;
-    GNUNET_CRYPTO_ecdsa_key_get_public (&mem->priv_key, &req->member_key);
+    GNUNET_CRYPTO_ecdsa_key_get_public (&mem->priv_key, &req->member_pub_key);
     if (0 < join_msg_size)
       memcpy (&req[1], join_msg, join_msg_size);
 
+    req->member_pub_key = mem->pub_key;
     req->purpose.size = htonl (msg_size
                                - sizeof (req->header)
                                - sizeof (req->reserved)
@@ -863,7 +1084,7 @@ static void
 client_send_join_decision (struct Member *mem,
                            const struct MulticastJoinDecisionMessageHeader *hdcsn)
 {
-  client_send_msg (&mem->grp, &hdcsn->header);
+  client_send_group (&mem->grp, &hdcsn->header);
 
   const struct MulticastJoinDecisionMessage *
     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
@@ -874,14 +1095,7 @@ client_send_join_decision (struct Member *mem,
     memcpy (mem->join_dcsn, dcsn, dcsn_size);
   }
   else
-  { /* Refused entry, disconnect clients. */
-    struct ClientList *cl = mem->grp.clients_head;
-    while (NULL != cl)
-    {
-      struct GNUNET_SERVER_Client *client = cl->client;
-      cl = cl->next;
-      GNUNET_SERVER_client_disconnect (client);
-    }
+  { /* Refused entry, but replay would be still possible for past members. */
   }
 }
 
@@ -915,7 +1129,7 @@ client_recv_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
   if (NULL != grp_mem)
   {
     struct GNUNET_HashCode member_key_hash;
-    GNUNET_CRYPTO_hash (&hdcsn->member_key, sizeof (hdcsn->member_key),
+    GNUNET_CRYPTO_hash (&hdcsn->member_pub_key, sizeof (hdcsn->member_pub_key),
                         &member_key_hash);
     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &member_key_hash);
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -943,8 +1157,7 @@ client_recv_multicast_message (void *cls, struct GNUNET_SERVER_Client *client,
 {
   struct Group *
     grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
-  const struct GNUNET_MULTICAST_MessageHeader *
-    msg = (const struct GNUNET_MULTICAST_MessageHeader *) m;
+  struct GNUNET_MULTICAST_MessageHeader *out;
   struct Origin *orig;
 
   if (NULL == grp)
@@ -956,21 +1169,27 @@ client_recv_multicast_message (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_assert (GNUNET_YES == grp->is_origin);
   orig = (struct Origin *) grp;
 
-  msg->fragment_id = GNUNET_htonll (++orig->max_fragment_id);
-  msg->purpose.size = htonl (ntohs (msg->header.size)
-                             - sizeof (msg->header)
-                             - sizeof (msg->hop_counter)
-                             - sizeof (msg->signature));
-  msg->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE);
-
-  if (GNUNET_OK != GNUNET_CRYPTO_eddsa_sign (&orig->priv_key, &msg->purpose,
-                                             &msg->signature))
+  /* FIXME: yucky, should use separate message structs for P2P and CS! */
+  out = (struct GNUNET_MULTICAST_MessageHeader *) GNUNET_copy_message (m);
+  out->fragment_id = GNUNET_htonll (++orig->max_fragment_id);
+  out->purpose.size = htonl (ntohs (out->header.size)
+                             - sizeof (out->header)
+                             - sizeof (out->hop_counter)
+                             - sizeof (out->signature));
+  out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE);
+
+  if (GNUNET_OK != GNUNET_CRYPTO_eddsa_sign (&orig->priv_key, &out->purpose,
+                                             &out->signature))
   {
     GNUNET_assert (0);
   }
 
-  client_send_all (&grp->pub_key_hash, m);
-  cadet_send_members (&grp->pub_key_hash, m);
+  client_send_all (&grp->pub_key_hash, &out->header);
+  if (0 == cadet_send_children (&grp->pub_key_hash, &out->header))
+  {
+    client_send_ack (&grp->pub_key_hash);
+  }
+  GNUNET_free (out);
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
@@ -985,9 +1204,7 @@ client_recv_multicast_request (void *cls, struct GNUNET_SERVER_Client *client,
 {
   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
   struct Member *mem;
-  const struct GNUNET_MULTICAST_RequestHeader *
-    req = (const struct GNUNET_MULTICAST_RequestHeader *) m;
-
+  struct GNUNET_MULTICAST_RequestHeader *out;
   if (NULL == grp)
   {
     GNUNET_break (0);
@@ -997,32 +1214,222 @@ client_recv_multicast_request (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_assert (GNUNET_NO == grp->is_origin);
   mem = (struct Member *) grp;
 
-  req->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
-  req->purpose.size = htonl (ntohs (req->header.size)
-                             - sizeof (req->header)
-                             - sizeof (req->member_key)
-                             - sizeof (req->signature));
-  req->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
-
-  if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &req->purpose,
-                                             &req->signature))
+  /* FIXME: yucky, should use separate message structs for P2P and CS! */
+  out = (struct GNUNET_MULTICAST_RequestHeader *) GNUNET_copy_message (m);
+  out->member_pub_key = mem->pub_key;
+  out->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
+  out->purpose.size = htonl (ntohs (out->header.size)
+                             - sizeof (out->header)
+                             - sizeof (out->member_pub_key)
+                             - sizeof (out->signature));
+  out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
+
+  if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &out->purpose,
+                                             &out->signature))
   {
     GNUNET_assert (0);
   }
 
-  if (0 == client_send_origin (&grp->pub_key_hash, m))
+  uint8_t send_ack = GNUNET_YES;
+  if (0 == client_send_origin (&grp->pub_key_hash, &out->header))
   { /* No local origins, send to remote origin */
     if (NULL != mem->origin_channel)
     {
-      cadet_send_msg (mem->origin_channel, m);
+      cadet_send_channel (mem->origin_channel, &out->header);
+      send_ack = GNUNET_NO;
     }
     else
     {
       /* FIXME: not yet connected to origin */
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+      GNUNET_free (out);
       return;
     }
   }
+  if (GNUNET_YES == send_ack)
+  {
+    client_send_ack (&grp->pub_key_hash);
+  }
+  GNUNET_free (out);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Incoming replay request from a client.
+ */
+static void
+client_recv_replay_request (void *cls, struct GNUNET_SERVER_Client *client,
+                            const struct GNUNET_MessageHeader *m)
+{
+  struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
+  struct Member *mem;
+  if (NULL == grp)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  GNUNET_assert (GNUNET_NO == grp->is_origin);
+  mem = (struct Member *) grp;
+
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
+                                                        &grp->pub_key_hash);
+  if (NULL == grp_replay_req)
+  {
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+    GNUNET_CONTAINER_multihashmap_put (replay_req_client,
+                                       &grp->pub_key_hash, grp_replay_req,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+  }
+  struct MulticastReplayRequestMessage *
+    rep = (struct MulticastReplayRequestMessage *) m;
+  struct GNUNET_HashCode key_hash;
+  replay_key_hash (rep->fragment_id, rep->message_id, rep->fragment_offset,
+                   rep->flags, &key_hash);
+  GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, client,
+                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+
+  if (0 == client_send_origin (&grp->pub_key_hash, m))
+  { /* No local origin, replay from remote members / origin. */
+    if (NULL != mem->origin_channel)
+    {
+      cadet_send_channel (mem->origin_channel, m);
+    }
+    else
+    {
+      /* FIXME: not yet connected to origin */
+      GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+      return;
+    }
+  }
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+static int
+cadet_send_replay_response_cb (void *cls,
+                               const struct GNUNET_HashCode *key_hash,
+                               void *value)
+{
+  struct Channel *chn = value;
+  struct GNUNET_MessageHeader *msg = cls;
+
+  cadet_send_channel (chn, msg);
+  return GNUNET_OK;
+}
+
+
+static int
+client_send_replay_response_cb (void *cls,
+                                const struct GNUNET_HashCode *key_hash,
+                                void *value)
+{
+  struct GNUNET_SERVER_Client *client = value;
+  struct GNUNET_MessageHeader *msg = cls;
+
+  client_send (client, msg);
+  return GNUNET_OK;
+}
+
+
+/**
+ * End of replay response from a client.
+ */
+static void
+client_recv_replay_response_end (void *cls, struct GNUNET_SERVER_Client *client,
+                                 const struct GNUNET_MessageHeader *m)
+{
+  struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
+  if (NULL == grp)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  struct MulticastReplayResponseMessage *
+    res = (struct MulticastReplayResponseMessage *) m;
+
+  struct GNUNET_HashCode key_hash;
+  replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
+                   res->flags, &key_hash);
+
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
+                                                                &grp->pub_key_hash);
+  if (NULL != grp_replay_req_cadet)
+  {
+    GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_cadet, &key_hash);
+  }
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
+                                                               &grp->pub_key_hash);
+  if (NULL != grp_replay_req_client)
+  {
+    GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_client, &key_hash);
+  }
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Incoming replay response from a client.
+ *
+ * Respond with a multicast message on success, or otherwise with an error code.
+ */
+static void
+client_recv_replay_response (void *cls, struct GNUNET_SERVER_Client *client,
+                             const struct GNUNET_MessageHeader *m)
+{
+  struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
+  if (NULL == grp)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  struct MulticastReplayResponseMessage *
+    res = (struct MulticastReplayResponseMessage *) m;
+
+  const struct GNUNET_MessageHeader *msg = m;
+  if (GNUNET_MULTICAST_REC_OK == res->error_code)
+  {
+    msg = (struct GNUNET_MessageHeader *) &res[1];
+  }
+
+  struct GNUNET_HashCode key_hash;
+  replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
+                   res->flags, &key_hash);
+
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
+                                                              &grp->pub_key_hash);
+  if (NULL != grp_replay_req_cadet)
+  {
+    GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_cadet, &key_hash,
+                                                cadet_send_replay_response_cb,
+                                                (void *) msg);
+  }
+  if (GNUNET_MULTICAST_REC_OK == res->error_code)
+  {
+    struct GNUNET_CONTAINER_MultiHashMap *
+      grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
+                                                                 &grp->pub_key_hash);
+    if (NULL != grp_replay_req_client)
+    {
+      GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_client, &key_hash,
+                                                  client_send_replay_response_cb,
+                                                  (void *) msg);
+    }
+  }
+  else
+  {
+    client_recv_replay_response_end (cls, client, m);
+    return;
+  }
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
@@ -1042,22 +1449,31 @@ client_notify_connect (void *cls, struct GNUNET_SERVER_Client *client)
  * Message handlers for the server.
  */
 static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
-  { &client_recv_origin_start, NULL,
+  { client_recv_origin_start, NULL,
     GNUNET_MESSAGE_TYPE_MULTICAST_ORIGIN_START, 0 },
 
-  { &client_recv_member_join, NULL,
+  { client_recv_member_join, NULL,
     GNUNET_MESSAGE_TYPE_MULTICAST_MEMBER_JOIN, 0 },
 
-  { &client_recv_join_decision, NULL,
+  { client_recv_join_decision, NULL,
     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION, 0 },
 
-  { &client_recv_multicast_message, NULL,
+  { client_recv_multicast_message, NULL,
     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
 
-  { &client_recv_multicast_request, NULL,
+  { client_recv_multicast_request, NULL,
     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
 
-  {NULL, NULL, 0, 0}
+  { client_recv_replay_request, NULL,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
+
+  { client_recv_replay_response, NULL,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
+
+  { client_recv_replay_response_end, NULL,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE_END, 0 },
+
+  { NULL, NULL, 0, 0 }
 };
 
 
@@ -1096,6 +1512,9 @@ cadet_notify_channel_end (void *cls,
         mem->origin_channel = NULL;
     }
   }
+
+  while (GNUNET_YES == replay_req_remove_cadet (chn));
+
   GNUNET_free (chn);
 }
 
@@ -1133,26 +1552,26 @@ cadet_recv_join_request (void *cls,
   if (GNUNET_OK !=
       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
                                   &req->purpose, &req->signature,
-                                  &req->member_key))
+                                  &req->member_pub_key))
   {
     GNUNET_break_op (0);
     return GNUNET_SYSERR;
   }
 
-  struct GNUNET_HashCode group_key_hash;
-  GNUNET_CRYPTO_hash (&req->group_key, sizeof (req->group_key), &group_key_hash);
+  struct GNUNET_HashCode group_pub_hash;
+  GNUNET_CRYPTO_hash (&req->group_pub_key, sizeof (req->group_pub_key), &group_pub_hash);
 
   struct Channel *chn = GNUNET_malloc (sizeof *chn);
   chn->channel = channel;
-  chn->group_key = req->group_key;
-  chn->group_key_hash = group_key_hash;
-  chn->member_key = req->member_key;
+  chn->group_pub_key = req->group_pub_key;
+  chn->group_pub_hash = group_pub_hash;
+  chn->member_pub_key = req->member_pub_key;
   chn->peer = req->peer;
   chn->join_status = JOIN_WAITING;
-  GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group_key_hash, chn,
+  GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group_pub_hash, chn,
                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 
-  client_send_all (&group_key_hash, m);
+  client_send_all (&group_pub_hash, m);
   return GNUNET_OK;
 }
 
@@ -1252,13 +1671,13 @@ cadet_recv_message (void *cls,
   if (GNUNET_OK !=
       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE,
                                   &msg->purpose, &msg->signature,
-                                  &chn->group_key))
+                                  &chn->group_pub_key))
   {
     GNUNET_break_op (0);
     return GNUNET_SYSERR;
   }
 
-  client_send_all (&chn->group_key_hash, m);
+  client_send_all (&chn->group_pub_hash, m);
   return GNUNET_OK;
 }
 
@@ -1288,7 +1707,7 @@ cadet_recv_request (void *cls,
   }
   if (ntohl (req->purpose.size) != (size
                                     - sizeof (req->header)
-                                    - sizeof (req->member_key)
+                                    - sizeof (req->member_pub_key)
                                     - sizeof (req->signature)))
   {
     GNUNET_break_op (0);
@@ -1297,13 +1716,72 @@ cadet_recv_request (void *cls,
   if (GNUNET_OK !=
       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
                                   &req->purpose, &req->signature,
-                                  &req->member_key))
+                                  &req->member_pub_key))
   {
     GNUNET_break_op (0);
     return GNUNET_SYSERR;
   }
 
-  client_send_origin (&chn->group_key_hash, m);
+  client_send_origin (&chn->group_pub_hash, m);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast replay request from CADET.
+ */
+int
+cadet_recv_replay_request (void *cls,
+                           struct GNUNET_CADET_Channel *channel,
+                           void **ctx,
+                           const struct GNUNET_MessageHeader *m)
+{
+  struct MulticastReplayRequestMessage rep;
+  uint16_t size = ntohs (m->size);
+  if (size < sizeof (rep))
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  struct Channel *chn = *ctx;
+
+  memcpy (&rep, m, sizeof (rep));
+  memcpy (&rep.member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key));
+
+  struct GNUNET_CONTAINER_MultiHashMap *
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
+                                                        &chn->grp->pub_key_hash);
+  if (NULL == grp_replay_req)
+  {
+    grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+    GNUNET_CONTAINER_multihashmap_put (replay_req_cadet,
+                                       &chn->grp->pub_key_hash, grp_replay_req,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+  }
+  struct GNUNET_HashCode key_hash;
+  replay_key_hash (rep.fragment_id, rep.message_id, rep.fragment_offset,
+                   rep.flags, &key_hash);
+  GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn,
+                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+
+  client_send_random (&chn->group_pub_hash, &rep.header);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast replay response from CADET.
+ */
+int
+cadet_recv_replay_response (void *cls,
+                            struct GNUNET_CADET_Channel *channel,
+                            void **ctx,
+                            const struct GNUNET_MessageHeader *m)
+{
+  struct Channel *chn = *ctx;
+
+  /* @todo FIXME: got replay error response, send request to other members */
+
   return GNUNET_OK;
 }
 
@@ -1312,9 +1790,21 @@ cadet_recv_request (void *cls,
  * Message handlers for CADET.
  */
 static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
-  { &cadet_recv_join_request, GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST, 0 },
-  { &cadet_recv_message, GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
-  { &cadet_recv_request, GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
+  { cadet_recv_join_request,
+    GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST, 0 },
+
+  { cadet_recv_message,
+    GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
+
+  { cadet_recv_request,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
+
+  { cadet_recv_replay_request,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
+
+  { cadet_recv_replay_response,
+    GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
+
   { NULL, 0, 0 }
 };
 
@@ -1339,6 +1829,8 @@ core_connected_cb  (void *cls, const struct GNUNET_PeerIdentity *my_identity)
   group_members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
   channels_in = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
   channels_out = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  replay_req_cadet = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+  replay_req_client = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
 
   cadet = GNUNET_CADET_connect (cfg, NULL,
                                 &cadet_notify_channel_new,
@@ -1347,10 +1839,10 @@ core_connected_cb  (void *cls, const struct GNUNET_PeerIdentity *my_identity)
 
   nc = GNUNET_SERVER_notification_context_create (server, 1);
   GNUNET_SERVER_add_handlers (server, server_handlers);
-  GNUNET_SERVER_disconnect_notify (server, &client_notify_disconnect, NULL);
-
-  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
-                                NULL);
+  GNUNET_SERVER_disconnect_notify (server,
+                                  &client_notify_disconnect, NULL);
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
+                                NULL);
 }
 
 
@@ -1362,7 +1854,8 @@ core_connected_cb  (void *cls, const struct GNUNET_PeerIdentity *my_identity)
  * @param cfg configuration to use
  */
 static void
-run (void *cls, struct GNUNET_SERVER_Handle *srv,
+run (void *cls,
+     struct GNUNET_SERVER_Handle *srv,
      const struct GNUNET_CONFIGURATION_Handle *c)
 {
   cfg = c;