-keep track of messages passed to mq
[oweals/gnunet.git] / src / sensor / gnunet-service-sensor_reporting.c
index 6267e9f2a971241e8e82e3197b467d7bba761172..4ea570ccd4e2a89899da29fb06f2a8b50bbde1b3 100644 (file)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     (C)
+     Copyright (C)
 
      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.
 */
 
 /**
 
 #define LOG(kind,...) GNUNET_log_from (kind, "sensor-reporting",__VA_ARGS__)
 
+/**
+ * Retry time when failing to connect to collection point
+ */
+#define CP_RETRY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1)
+
+
+/**
+ * When we are still generating a proof-of-work and we need to send an anomaly
+ * report, we queue them until the generation is complete
+ */
+struct AnomalyReportingQueueItem
+{
+
+  /**
+   * DLL
+   */
+  struct AnomalyReportingQueueItem *prev;
+
+  /**
+   * DLL
+   */
+  struct AnomalyReportingQueueItem *next;
+
+  /**
+   * Message queue belonging to the peer that is the destination of the report
+   */
+  struct GNUNET_MQ_Handle *dest_mq;
+
+  /**
+   * Report type
+   */
+  int type;
+
+};
 
 struct AnomalyInfo
 {
@@ -62,6 +96,26 @@ struct AnomalyInfo
    */
   struct GNUNET_CONTAINER_MultiPeerMap *anomalous_neighbors;
 
+  /**
+   * Report block with proof-of-work and signature
+   */
+  struct GNUNET_SENSOR_crypto_pow_block *report_block;
+
+  /**
+   * Context of an operation creating pow and signature
+   */
+  struct GNUNET_SENSOR_crypto_pow_context *report_creation_cx;
+
+  /**
+   * Head of the queue of pending report destinations
+   */
+  struct AnomalyReportingQueueItem *reporting_queue_head;
+
+  /**
+   * Head of the queue of pending report destinations
+   */
+  struct AnomalyReportingQueueItem *reporting_queue_tail;
+
 };
 
 struct ValueInfo
@@ -108,9 +162,9 @@ struct ValueInfo
   struct GNUNET_PEERSTORE_WatchContext *wc;
 
   /**
-   * Collection point reporting task (or #GNUNET_SCHEDULER_NO_TASK)
+   * Collection point reporting task (or NULL)
    */
-  GNUNET_SCHEDULER_TaskIdentifier reporting_task;
+  struct GNUNET_SCHEDULER_Task *reporting_task;
 
 };
 
@@ -175,6 +229,16 @@ struct CadetPeer
    */
   struct GNUNET_MQ_Handle *mq;
 
+  /**
+   * CADET transmit handle
+   */
+  struct GNUNET_CADET_TransmitHandle *th;
+
+  /**
+   * Task used to try reconnection to collection point after failure
+   */
+  struct GNUNET_SCHEDULER_Task * reconnect_task;
+
   /**
    * Are we currently destroying the channel and its context?
    */
@@ -213,6 +277,11 @@ static struct GNUNET_CADET_Handle *cadet;
  */
 static struct GNUNET_PeerIdentity mypeerid;
 
+/**
+ * My private key
+ */
+static struct GNUNET_CRYPTO_EddsaPrivateKey *private_key;
+
 /**
  * Head of DLL of anomaly info structs
  */
@@ -263,6 +332,18 @@ static int module_running = GNUNET_NO;
  */
 static int neighborhood;
 
+/**
+ * Parameter that defines the complexity of the proof-of-work
+ */
+static long long unsigned int pow_matching_bits;
+
+
+
+/**
+ * Try reconnecting to collection point and send last queued message
+ */
+static void
+cp_reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 
 
 /******************************************************************************/
@@ -277,8 +358,31 @@ static int neighborhood;
 static void
 destroy_anomaly_info (struct AnomalyInfo *ai)
 {
+  struct AnomalyReportingQueueItem *ar_item;
+
+  ar_item = ai->reporting_queue_head;
+  while (NULL != ar_item)
+  {
+    GNUNET_CONTAINER_DLL_remove (ai->reporting_queue_head,
+                                 ai->reporting_queue_tail, ar_item);
+    GNUNET_free (ar_item);
+    ar_item = ai->reporting_queue_head;
+  }
+  if (NULL != ai->report_creation_cx)
+  {
+    GNUNET_SENSOR_crypto_pow_sign_cancel (ai->report_creation_cx);
+    ai->report_creation_cx = NULL;
+  }
+  if (NULL != ai->report_block)
+  {
+    GNUNET_free (ai->report_block);
+    ai->report_block = NULL;
+  }
   if (NULL != ai->anomalous_neighbors)
+  {
     GNUNET_CONTAINER_multipeermap_destroy (ai->anomalous_neighbors);
+    ai->anomalous_neighbors = NULL;
+  }
   GNUNET_free (ai);
 }
 
@@ -296,10 +400,10 @@ destroy_value_info (struct ValueInfo *vi)
     GNUNET_PEERSTORE_watch_cancel (vi->wc);
     vi->wc = NULL;
   }
-  if (GNUNET_SCHEDULER_NO_TASK != vi->reporting_task)
+  if (NULL != vi->reporting_task)
   {
     GNUNET_SCHEDULER_cancel (vi->reporting_task);
-    vi->reporting_task = GNUNET_SCHEDULER_NO_TASK;
+    vi->reporting_task = NULL;
   }
   if (NULL != vi->last_value)
   {
@@ -319,20 +423,33 @@ static void
 destroy_core_peer (struct CorePeer *corep)
 {
   struct AnomalyInfo *ai;
+  struct AnomalyReportingQueueItem *ar_item;
 
-  if (NULL != corep->mq)
-  {
-    GNUNET_MQ_destroy (corep->mq);
-    corep->mq = NULL;
-  }
   ai = ai_head;
   while (NULL != ai)
   {
     GNUNET_assert (NULL != ai->anomalous_neighbors);
     GNUNET_CONTAINER_multipeermap_remove_all (ai->anomalous_neighbors,
                                               corep->peer_id);
+    /* Remove the core peer from any reporting queues */
+    ar_item = ai->reporting_queue_head;
+    while (NULL != ar_item)
+    {
+      if (ar_item->dest_mq == corep->mq)
+      {
+        GNUNET_CONTAINER_DLL_remove (ai->reporting_queue_head,
+                                     ai->reporting_queue_tail, ar_item);
+        break;
+      }
+      ar_item = ar_item->next;
+    }
     ai = ai->next;
   }
+  if (NULL != corep->mq)
+  {
+    GNUNET_MQ_destroy (corep->mq);
+    corep->mq = NULL;
+  }
   GNUNET_free (corep);
 }
 
@@ -346,6 +463,11 @@ static void
 destroy_cadet_peer (struct CadetPeer *cadetp)
 {
   cadetp->destroying = GNUNET_YES;
+  if (NULL != cadetp->reconnect_task)
+  {
+    GNUNET_SCHEDULER_cancel (cadetp->reconnect_task);
+    cadetp->reconnect_task = NULL;
+  }
   if (NULL != cadetp->mq)
   {
     GNUNET_MQ_destroy (cadetp->mq);
@@ -453,6 +575,142 @@ get_anomaly_info_by_sensor (struct GNUNET_SENSOR_SensorInfo *sensor)
 }
 
 
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+cp_mq_ntr (void *cls, size_t size, void *buf)
+{
+  struct CadetPeer *cadetp = cls;
+  const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (cadetp->mq);
+  uint16_t msize;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "cp_mq_ntr()\n");
+  cadetp->th = NULL;
+  if (NULL == buf)
+  {
+    LOG (GNUNET_ERROR_TYPE_INFO,
+         "Sending anomaly report to collection point failed."
+         " Retrying connection in %s.\n",
+         GNUNET_STRINGS_relative_time_to_string (CP_RETRY, GNUNET_NO));
+    cadetp->reconnect_task =
+        GNUNET_SCHEDULER_add_delayed (CP_RETRY, &cp_reconnect, cadetp);
+    return 0;
+  }
+  msize = ntohs (msg->size);
+  GNUNET_assert (msize <= size);
+  memcpy (buf, msg, msize);
+  GNUNET_MQ_impl_send_continue (cadetp->mq);
+  return msize;
+}
+
+
+/**
+ * Try reconnecting to collection point and send last queued message
+ */
+static void
+cp_reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct CadetPeer *cadetp = cls;
+  const struct GNUNET_MessageHeader *msg;
+
+  LOG (GNUNET_ERROR_TYPE_INFO,
+       "Retrying connection to collection point `%s'.\n",
+       GNUNET_i2s (&cadetp->peer_id));
+  cadetp->reconnect_task = NULL;
+  GNUNET_assert (NULL == cadetp->channel);
+  cadetp->channel =
+      GNUNET_CADET_channel_create (cadet, cadetp, &cadetp->peer_id,
+                                   GNUNET_APPLICATION_TYPE_SENSORDASHBOARD,
+                                   GNUNET_CADET_OPTION_RELIABLE);
+  msg = GNUNET_MQ_impl_current (cadetp->mq);
+  cadetp->th =
+      GNUNET_CADET_notify_transmit_ready (cadetp->channel, GNUNET_NO,
+                                          GNUNET_TIME_UNIT_FOREVER_REL,
+                                          ntohs (msg->size), cp_mq_ntr, cadetp);
+}
+
+
+/**
+ * Signature of functions implementing the
+ * sending functionality of a message queue.
+ *
+ * @param mq the message queue
+ * @param msg the message to send
+ * @param impl_state state of the implementation
+ */
+static void
+cp_mq_send_impl (struct GNUNET_MQ_Handle *mq,
+                 const struct GNUNET_MessageHeader *msg, void *impl_state)
+{
+  struct CadetPeer *cadetp = impl_state;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "cp_mq_send_impl()\n");
+  GNUNET_assert (NULL == cadetp->th);
+  if (NULL == cadetp->channel)
+  {
+    LOG (GNUNET_ERROR_TYPE_INFO,
+         "Sending anomaly report to collection point failed."
+         " Retrying connection in %s.\n",
+         GNUNET_STRINGS_relative_time_to_string (CP_RETRY, GNUNET_NO));
+    cadetp->reconnect_task =
+        GNUNET_SCHEDULER_add_delayed (CP_RETRY, &cp_reconnect, cadetp);
+    return;
+  }
+  cadetp->th =
+      GNUNET_CADET_notify_transmit_ready (cadetp->channel, GNUNET_NO,
+                                          GNUNET_TIME_UNIT_FOREVER_REL,
+                                          ntohs (msg->size), cp_mq_ntr, cadetp);
+}
+
+
+/**
+ * Signature of functions implementing the
+ * destruction of a message queue.
+ * Implementations must not free 'mq', but should
+ * take care of 'impl_state'.
+ *
+ * @param mq the message queue to destroy
+ * @param impl_state state of the implementation
+ */
+static void
+cp_mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
+{
+  struct CadetPeer *cp = impl_state;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "cp_mq_destroy_impl()\n");
+  if (NULL != cp->th)
+  {
+    GNUNET_CADET_notify_transmit_ready_cancel (cp->th);
+    cp->th = NULL;
+  }
+}
+
+
+/**
+ * Create the message queue used to send messages to a collection point.
+ * This will be used to make sure that the message are queued even if the
+ * connection to the collection point can not be established at the moment.
+ *
+ * @param cp CadetPeer information struct
+ * @return Message queue handle
+ */
+static struct GNUNET_MQ_Handle *
+cp_mq_create (struct CadetPeer *cp)
+{
+  return GNUNET_MQ_queue_for_callbacks (cp_mq_send_impl, cp_mq_destroy_impl,
+                                        NULL, cp, NULL, NULL, NULL);
+}
+
+
 /**
  * Returns context of a connected CADET peer.
  * Creates it first if didn't exist before.
@@ -481,27 +739,146 @@ get_cadet_peer (struct GNUNET_PeerIdentity pid)
       GNUNET_CADET_channel_create (cadet, cadetp, &pid,
                                    GNUNET_APPLICATION_TYPE_SENSORDASHBOARD,
                                    GNUNET_CADET_OPTION_RELIABLE);
-  cadetp->mq = GNUNET_CADET_mq_create (cadetp->channel);
+  cadetp->mq = cp_mq_create (cadetp);
+  cadetp->reconnect_task = NULL;
   GNUNET_CONTAINER_DLL_insert (cadetp_head, cadetp_tail, cadetp);
   return cadetp;
 }
 
 
 /**
- * Create an anomaly report message from a given anomaly info struct inside a
- * MQ envelope.
+ * This function is called only when we have a block ready and want to send it
+ * to the given peer (represented by its message queue)
  *
- * @param ai Anomaly info struct to use
+ * @param mq Message queue to put the message in
+ * @param ai Anomaly info to report
  * @param type Message type
- * @return Envelope with message
  */
-static struct GNUNET_MQ_Envelope *
-create_anomaly_report_message (struct AnomalyInfo *ai, int type)
+static void
+do_send_anomaly_report (struct GNUNET_MQ_Handle *mq, struct AnomalyInfo *ai,
+                        int type)
 {
-  struct GNUNET_SENSOR_AnomalyReportMessage *arm;
+  struct GNUNET_MessageHeader *msg;
   struct GNUNET_MQ_Envelope *ev;
+  size_t block_size;
+
+  GNUNET_assert (NULL != ai->report_block);
+  block_size =
+      sizeof (struct GNUNET_SENSOR_crypto_pow_block) +
+      ai->report_block->msg_size;
+  ev = GNUNET_MQ_msg_header_extra (msg, block_size, type);
+  memcpy (&msg[1], ai->report_block, block_size);
+  GNUNET_MQ_send (mq, ev);
+}
+
+
+/**
+ * Check if we have signed and proof-of-work block ready.
+ * If yes, we send the report directly, if no, we enqueue the reporting until
+ * the block is ready.
+ *
+ * @param mq Message queue to put the message in
+ * @param ai Anomaly info to report
+ * @param p2p Is the report sent to a neighboring peer
+ */
+static void
+send_anomaly_report (struct GNUNET_MQ_Handle *mq, struct AnomalyInfo *ai,
+                     int p2p)
+{
+  struct AnomalyReportingQueueItem *ar_item;
+  int type;
+
+  type =
+      (GNUNET_YES ==
+       p2p) ? GNUNET_MESSAGE_TYPE_SENSOR_ANOMALY_REPORT_P2P :
+      GNUNET_MESSAGE_TYPE_SENSOR_ANOMALY_REPORT;
+  if (NULL == ai->report_block)
+  {
+    ar_item = GNUNET_new (struct AnomalyReportingQueueItem);
+
+    ar_item->dest_mq = mq;
+    ar_item->type = type;
+    GNUNET_CONTAINER_DLL_insert_tail (ai->reporting_queue_head,
+                                      ai->reporting_queue_tail, ar_item);
+  }
+  else
+  {
+    do_send_anomaly_report (mq, ai, type);
+  }
+}
+
+
+/**
+ * Callback when the crypto module finished created proof-of-work and signature
+ * for an anomaly report.
+ *
+ * @param cls Closure, a `struct AnomalyInfo *`
+ * @param block The resulting block, NULL on error
+ */
+static void
+report_creation_cb (void *cls, struct GNUNET_SENSOR_crypto_pow_block *block)
+{
+  struct AnomalyInfo *ai = cls;
+  struct AnomalyReportingQueueItem *ar_item;
+
+  ai->report_creation_cx = NULL;
+  if (NULL != ai->report_block)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         _("Double creation of proof-of-work, this should not happen.\n"));
+    return;
+  }
+  if (NULL == block)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         _("Failed to create pow and signature block.\n"));
+    return;
+  }
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Anomaly report POW block ready.\n");
+  ai->report_block =
+      GNUNET_memdup (block,
+                     sizeof (struct GNUNET_SENSOR_crypto_pow_block) +
+                     block->msg_size);
+  ar_item = ai->reporting_queue_head;
+  while (NULL != ar_item)
+  {
+    GNUNET_CONTAINER_DLL_remove (ai->reporting_queue_head,
+                                 ai->reporting_queue_tail, ar_item);
+    do_send_anomaly_report (ar_item->dest_mq, ai, ar_item->type);
+    GNUNET_free (ar_item);
+    ar_item = ai->reporting_queue_head;
+  }
+}
+
+
+/**
+ * When a change to the anomaly info of a sensor is done, this function should
+ * be called to create the message, its proof-of-work and signuature ready to
+ * be sent to other peers or collection point.
+ *
+ * @param ai Anomaly Info struct
+ */
+static void
+update_anomaly_report_pow_block (struct AnomalyInfo *ai)
+{
+  struct GNUNET_SENSOR_AnomalyReportMessage *arm;
+  struct GNUNET_TIME_Absolute timestamp;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Updating anomaly report POW block due to data change.\n");
+  if (NULL != ai->report_block)
+  {
+    GNUNET_free (ai->report_block);
+    ai->report_block = NULL;
+  }
+  if (NULL != ai->report_creation_cx)
+  {
+    /* If a creation is already running, cancel it because the data changed */
+    GNUNET_SENSOR_crypto_pow_sign_cancel (ai->report_creation_cx);
+    ai->report_creation_cx = NULL;
+  }
+  arm = GNUNET_new (struct GNUNET_SENSOR_AnomalyReportMessage);
 
-  ev = GNUNET_MQ_msg (arm, type);
   GNUNET_CRYPTO_hash (ai->sensor->name, strlen (ai->sensor->name) + 1,
                       &arm->sensorname_hash);
   arm->sensorversion_major = htons (ai->sensor->version_major);
@@ -512,7 +889,15 @@ create_anomaly_report_message (struct AnomalyInfo *ai, int type)
        neighborhood) ? 0 : ((float)
                             GNUNET_CONTAINER_multipeermap_size
                             (ai->anomalous_neighbors)) / neighborhood;
-  return ev;
+  timestamp = GNUNET_TIME_absolute_get ();
+  ai->report_creation_cx =
+      GNUNET_SENSOR_crypto_pow_sign (arm,
+                                     sizeof (struct
+                                             GNUNET_SENSOR_AnomalyReportMessage),
+                                     &timestamp, &mypeerid.public_key,
+                                     private_key, pow_matching_bits,
+                                     &report_creation_cb, ai);
+  GNUNET_free (arm);
 }
 
 
@@ -542,29 +927,6 @@ create_value_message (struct ValueInfo *vi)
 }
 
 
-/**
- * Send given anomaly info report by putting it in the given message queue.
- *
- * @param mq Message queue to put the message in
- * @param ai Anomaly info to report
- * @param p2p Is the report sent to a neighboring peer
- */
-static void
-send_anomaly_report (struct GNUNET_MQ_Handle *mq, struct AnomalyInfo *ai,
-                     int p2p)
-{
-  struct GNUNET_MQ_Envelope *ev;
-  int type;
-
-  type =
-      (GNUNET_YES ==
-       p2p) ? GNUNET_MESSAGE_TYPE_SENSOR_ANOMALY_REPORT_P2P :
-      GNUNET_MESSAGE_TYPE_SENSOR_ANOMALY_REPORT;
-  ev = create_anomaly_report_message (ai, type);
-  GNUNET_MQ_send (mq, ev);
-}
-
-
 /******************************************************************************/
 /***************************      CORE Handlers     ***************************/
 /******************************************************************************/
@@ -583,6 +945,7 @@ static int
 handle_anomaly_report (void *cls, const struct GNUNET_PeerIdentity *other,
                        const struct GNUNET_MessageHeader *message)
 {
+  struct GNUNET_SENSOR_crypto_pow_block *report_block;
   struct GNUNET_SENSOR_AnomalyReportMessage *arm;
   struct GNUNET_SENSOR_SensorInfo *sensor;
   struct AnomalyInfo *my_anomaly_info;
@@ -590,7 +953,21 @@ handle_anomaly_report (void *cls, const struct GNUNET_PeerIdentity *other,
   int peer_anomalous;
   int peer_in_anomalous_list;
 
-  arm = (struct GNUNET_SENSOR_AnomalyReportMessage *) message;
+  /* Verify proof-of-work, signature and extract report message */
+  report_block = (struct GNUNET_SENSOR_crypto_pow_block *) &message[1];
+  if (sizeof (struct GNUNET_SENSOR_AnomalyReportMessage) !=
+      GNUNET_SENSOR_crypto_verify_pow_sign (report_block, pow_matching_bits,
+                                            (struct GNUNET_CRYPTO_EddsaPublicKey
+                                             *) &other->public_key,
+                                            (void **) &arm))
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Received invalid anomaly report from peer `%s'.\n",
+         GNUNET_i2s (other));
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  /* Now we parse the content of the message */
   sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &arm->sensorname_hash);
   if (NULL == sensor ||
       sensor->version_major != ntohs (arm->sensorversion_major) ||
@@ -607,6 +984,9 @@ handle_anomaly_report (void *cls, const struct GNUNET_PeerIdentity *other,
       GNUNET_CONTAINER_multipeermap_contains
       (my_anomaly_info->anomalous_neighbors, other);
   peer_anomalous = ntohs (arm->anomalous);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received an anomaly update from neighbour `%s' (%d).\n",
+       GNUNET_i2s (other), peer_anomalous);
   if (GNUNET_YES == peer_anomalous)
   {
     if (GNUNET_YES == peer_in_anomalous_list)   /* repeated positive report */
@@ -624,6 +1004,8 @@ handle_anomaly_report (void *cls, const struct GNUNET_PeerIdentity *other,
       GNUNET_CONTAINER_multipeermap_remove_all
           (my_anomaly_info->anomalous_neighbors, other);
   }
+  /* This is important to create an updated block since the data changed */
+  update_anomaly_report_pow_block (my_anomaly_info);
   /* Send anomaly update to collection point only if I have the same anomaly */
   if (GNUNET_YES == my_anomaly_info->anomalous &&
       NULL != sensor->collection_point &&
@@ -650,10 +1032,12 @@ handle_anomaly_report (void *cls, const struct GNUNET_PeerIdentity *other,
  * @param cls Closure, ValueInfo struct related to the sensor we are watching
  * @param record PEERSTORE new record, NULL if error
  * @param emsg Error message, NULL if no error
- * @return GNUNET_YES to continue watching
+ * @return #GNUNET_YES to continue watching
  */
 static int
-value_watch_cb (void *cls, struct GNUNET_PEERSTORE_Record *record, char *emsg)
+value_watch_cb (void *cls,
+                const struct GNUNET_PEERSTORE_Record *record,
+                const char *emsg)
 {
   struct ValueInfo *vi = cls;
 
@@ -804,9 +1188,12 @@ cadet_channel_destroyed (void *cls, const struct GNUNET_CADET_Channel *channel,
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "CADET channel was destroyed by remote peer `%s' or failed to start.\n",
        GNUNET_i2s (&cadetp->peer_id));
-  GNUNET_CONTAINER_DLL_remove (cadetp_head, cadetp_tail, cadetp);
+  if (NULL != cadetp->th)
+  {
+    GNUNET_CADET_notify_transmit_ready_cancel (cadetp->th);
+    cadetp->th = NULL;
+  }
   cadetp->channel = NULL;
-  destroy_cadet_peer (cadetp);
 }
 
 
@@ -836,6 +1223,8 @@ SENSOR_reporting_anomaly_update (struct GNUNET_SENSOR_SensorInfo *sensor,
   ai = get_anomaly_info_by_sensor (sensor);
   GNUNET_assert (NULL != ai);
   ai->anomalous = anomalous;
+  /* This is important to create an updated block since the data changed */
+  update_anomaly_report_pow_block (ai);
   /* Report change to all neighbors */
   corep = corep_head;
   while (NULL != corep)
@@ -891,6 +1280,12 @@ report_value (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
        "Now trying to report last seen value of `%s' to collection point.\n",
        sensor->name);
   cadetp = get_cadet_peer (*sensor->collection_point);
+  if (NULL == cadetp->channel)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Trying to send value to collection point but connection failed, discarding.\n");
+    return;
+  }
   ev = create_value_message (vi);
   GNUNET_MQ_send (cadetp->mq, ev);
   vi->last_value_reported = GNUNET_YES;
@@ -925,6 +1320,8 @@ init_sensor_reporting (void *cls, const struct GNUNET_HashCode *key,
   ai->anomalous = GNUNET_NO;
   ai->anomalous_neighbors =
       GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
+  ai->report_block = NULL;
+  ai->report_creation_cx = NULL;
   GNUNET_CONTAINER_DLL_insert (ai_head, ai_tail, ai);
   /* Create sensor value info context (if needed to be reported) */
   if (NULL == sensor->collection_point || GNUNET_NO == sensor->report_values)
@@ -963,6 +1360,8 @@ SENSOR_reporting_start (const struct GNUNET_CONFIGURATION_Handle *c,
 {
   static struct GNUNET_CORE_MessageHandler core_handlers[] = {
     {&handle_anomaly_report, GNUNET_MESSAGE_TYPE_SENSOR_ANOMALY_REPORT_P2P,
+     sizeof (struct GNUNET_MessageHeader) +
+     sizeof (struct GNUNET_SENSOR_crypto_pow_block) +
      sizeof (struct GNUNET_SENSOR_AnomalyReportMessage)},
     {NULL, 0, 0}
   };
@@ -974,6 +1373,23 @@ SENSOR_reporting_start (const struct GNUNET_CONFIGURATION_Handle *c,
   GNUNET_assert (NULL != s);
   sensors = s;
   cfg = c;
+  if (GNUNET_OK !=
+      GNUNET_CONFIGURATION_get_value_number (cfg, "sensor-reporting",
+                                             "POW_MATCHING_BITS",
+                                             &pow_matching_bits))
+  {
+    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, "sensor-reporting",
+                               "POW_MATCHING_BITS");
+    SENSOR_reporting_stop ();
+    return GNUNET_SYSERR;
+  }
+  if (pow_matching_bits > sizeof (struct GNUNET_HashCode))
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, "Matching bits value too large (%d > %d).\n",
+         pow_matching_bits, sizeof (struct GNUNET_HashCode));
+    SENSOR_reporting_stop ();
+    return GNUNET_SYSERR;
+  }
   /* Connect to PEERSTORE */
   peerstore = GNUNET_PEERSTORE_connect (cfg);
   if (NULL == peerstore)
@@ -1004,6 +1420,13 @@ SENSOR_reporting_start (const struct GNUNET_CONFIGURATION_Handle *c,
     SENSOR_reporting_stop ();
     return GNUNET_SYSERR;
   }
+  private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (cfg);
+  if (NULL == private_key)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, _("Failed to load my private key.\n"));
+    SENSOR_reporting_stop ();
+    return GNUNET_SYSERR;
+  }
   GNUNET_CRYPTO_get_peer_identity (cfg, &mypeerid);
   GNUNET_CONTAINER_multihashmap_iterate (sensors, &init_sensor_reporting, NULL);
   neighborhood = 0;