psyc, psycstore: retrieve state and history

5 files changed, 922 insertions, 171 deletions

diff --git a/src/psyc/Makefile.am b/src/psyc/Makefile.am
index b18605ab9..fb4341751 100644
--- a/src/psyc/Makefile.am
+++ b/src/psyc/Makefile.am
@@ -86,12 +86,14 @@ test_psyc_LDADD = \
   libgnunetpsyc.la \
   libgnunetpsycutil.la \
   $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/core/libgnunetcore.la \
   $(top_builddir)/src/env/libgnunetenv.la \
   $(top_builddir)/src/util/libgnunetutil.la
 test_psyc_DEPENDENCIES = \
   libgnunetpsyc.la \
   libgnunetpsycutil.la \
   $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/core/libgnunetcore.la \
   $(top_builddir)/src/env/libgnunetenv.la \
   $(top_builddir)/src/util/libgnunetutil.la
 
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index e7020bc69..3adc34d2a 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -402,6 +402,14 @@ struct Slave
 };
 
 
+struct OperationClosure
+{
+  struct GNUNET_SERVER_Client *client;
+  struct Channel *chn;
+  uint64_t op_id;
+};
+
+
 static void
 transmit_message (struct Channel *chn);
 
@@ -585,6 +593,46 @@ client_send_msg (const struct Channel *chn,
 
 
 /**
+ * Send a result code back to the client.
+ *
+ * @param client
+ *        Client that should receive the result code.
+ * @param result_code
+ *        Code to transmit.
+ * @param op_id
+ *        Operation ID in network byte order.
+ * @param err_msg
+ *        Error message to include (or NULL for none).
+ */
+static void
+client_send_result (struct GNUNET_SERVER_Client *client, uint64_t op_id,
+                    int64_t result_code, const char *err_msg)
+{
+  struct OperationResult *res;
+  size_t err_len = 0; // FIXME: maximum length
+
+  if (NULL != err_msg)
+    err_len = strlen (err_msg) + 1;
+  res = GNUNET_malloc (sizeof (struct OperationResult) + err_len);
+  res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE);
+  res->header.size = htons (sizeof (struct OperationResult) + err_len);
+  res->result_code = GNUNET_htonll (result_code + INT64_MAX + 1);
+  res->op_id = op_id;
+  if (0 < err_len)
+    memcpy (&res[1], err_msg, err_len);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Sending result to client for operation #%" PRIu64 ": "
+              "%" PRId64 " (%s)\n",
+              client, GNUNET_ntohll (op_id), result_code, err_msg);
+
+  GNUNET_SERVER_notification_context_add (nc, client);
+  GNUNET_SERVER_notification_context_unicast (nc, client, &res->header,
+                                              GNUNET_NO);
+  GNUNET_free (res);
+}
+
+
+/**
  * Closure for join_mem_test_cb()
  */
 struct JoinMemTestClosure
@@ -799,7 +847,8 @@ mcast_recv_replay_fragment (void *cls,
 
 {
   struct Channel *chn = cls;
-  GNUNET_PSYCSTORE_fragment_get (store, &chn->pub_key, slave_key, fragment_id,
+  GNUNET_PSYCSTORE_fragment_get (store, &chn->pub_key, slave_key,
+                                 fragment_id, fragment_id,
                                  &store_recv_fragment_replay,
                                  &store_recv_fragment_replay_result, rh);
 }
@@ -817,7 +866,8 @@ mcast_recv_replay_message (void *cls,
                            struct GNUNET_MULTICAST_ReplayHandle *rh)
 {
   struct Channel *chn = cls;
-  GNUNET_PSYCSTORE_message_get (store, &chn->pub_key, slave_key, message_id,
+  GNUNET_PSYCSTORE_message_get (store, &chn->pub_key, slave_key,
+                                message_id, message_id,
                                 &store_recv_fragment_replay,
                                 &store_recv_fragment_replay_result, rh);
 }
@@ -865,7 +915,8 @@ hash_key_from_hll (struct GNUNET_HashCode *key, uint64_t n)
  */
 static void
 client_send_mcast_msg (struct Channel *chn,
-                       const struct GNUNET_MULTICAST_MessageHeader *mmsg)
+                       const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                       uint32_t flags)
 {
   struct GNUNET_PSYC_MessageHeader *pmsg;
   uint16_t size = ntohs (mmsg->header.size);
@@ -882,6 +933,7 @@ client_send_mcast_msg (struct Channel *chn,
   pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
   pmsg->message_id = mmsg->message_id;
   pmsg->fragment_offset = mmsg->fragment_offset;
+  pmsg->flags = htonl (flags);
 
   memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
   client_send_msg (chn, &pmsg->header);
@@ -1111,7 +1163,7 @@ fragment_queue_run (struct Channel *chn, uint64_t msg_id,
     {
       if (GNUNET_NO == drop)
       {
-        client_send_mcast_msg (chn, cache_entry->mmsg);
+        client_send_mcast_msg (chn, cache_entry->mmsg, 0);
       }
       if (cache_entry->ref_count <= 1)
       {
@@ -1375,10 +1427,10 @@ store_recv_master_counters (void *cls, int result, uint64_t max_fragment_id,
   struct Channel *chn = &mst->chn;
   chn->store_op = NULL;
 
-  struct CountersResult res;
+  struct GNUNET_PSYC_CountersResultMessage res;
   res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
   res.header.size = htons (sizeof (res));
-  res.result_code = htonl (result);
+  res.result_code = htonl (result - INT32_MIN);
   res.max_message_id = GNUNET_htonll (max_message_id);
 
   if (GNUNET_OK == result || GNUNET_NO == result)
@@ -1421,10 +1473,10 @@ store_recv_slave_counters (void *cls, int result, uint64_t max_fragment_id,
   struct Channel *chn = &slv->chn;
   chn->store_op = NULL;
 
-  struct CountersResult res;
+  struct GNUNET_PSYC_CountersResultMessage res;
   res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
   res.header.size = htons (sizeof (res));
-  res.result_code = htonl (result);
+  res.result_code = htonl (result - INT32_MIN);
   res.max_message_id = GNUNET_htonll (max_message_id);
 
   if (GNUNET_OK == result || GNUNET_NO == result)
@@ -1511,10 +1563,10 @@ client_recv_master_start (void *cls, struct GNUNET_SERVER_Client *client,
   {
     chn = &mst->chn;
 
-    struct CountersResult res;
+    struct GNUNET_PSYC_CountersResultMessage res;
     res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
     res.header.size = htons (sizeof (res));
-    res.result_code = htonl (GNUNET_OK);
+    res.result_code = htonl ((uint32_t) GNUNET_OK + INT32_MIN);
     res.max_message_id = GNUNET_htonll (mst->max_message_id);
 
     GNUNET_SERVER_notification_context_add (nc, client);
@@ -1621,10 +1673,10 @@ client_recv_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
   {
     chn = &slv->chn;
 
-    struct CountersResult res;
+    struct GNUNET_PSYC_CountersResultMessage res;
     res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
     res.header.size = htons (sizeof (res));
-    res.result_code = htonl (GNUNET_OK);
+    res.result_code = htonl ((uint32_t) GNUNET_OK - INT32_MIN);
     res.max_message_id = GNUNET_htonll (chn->max_message_id);
 
     GNUNET_SERVER_notification_context_add (nc, client);
@@ -2047,17 +2099,26 @@ client_recv_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
 };
 
 
+struct MembershipStoreClosure
+{
+  struct GNUNET_SERVER_Client *client;
+  struct Channel *chn;
+  uint64_t op_id;
+};
+
+
 /**
  * Received result of GNUNET_PSYCSTORE_membership_store()
  */
 static void
 store_recv_membership_store_result (void *cls, int64_t result, const char *err_msg)
 {
-  struct GNUNET_MULTICAST_MembershipTestHandle *mth = cls;
+  struct MembershipStoreClosure *mcls = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "%p GNUNET_PSYCSTORE_membership_store() returned %" PRId64 " (%s)\n",
-              mth, result, err_msg);
-  /* FIXME: send result to client */
+              mcls->chn, result, err_msg);
+
+  client_send_result (mcls->client, mcls->op_id, result, err_msg);
 }
 
 
@@ -2075,6 +2136,11 @@ client_recv_membership_store (void *cls, struct GNUNET_SERVER_Client *client,
   const struct ChannelMembershipStoreRequest *
     req = (const struct ChannelMembershipStoreRequest *) msg;
 
+  struct MembershipStoreClosure *mcls = GNUNET_malloc (sizeof (*mcls));
+  mcls->client = client;
+  mcls->chn = chn;
+  mcls->op_id = req->op_id;
+
   uint64_t announced_at = GNUNET_ntohll (req->announced_at);
   uint64_t effective_since = GNUNET_ntohll (req->effective_since);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -2086,19 +2152,138 @@ client_recv_membership_store (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_PSYCSTORE_membership_store (store, &chn->pub_key, &req->slave_key,
                                      req->did_join, announced_at, effective_since,
                                      0, /* FIXME: group_generation */
-                                     &store_recv_membership_store_result, chn);
+                                     &store_recv_membership_store_result, mcls);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
+static int
+store_recv_fragment_history (void *cls,
+                             struct GNUNET_MULTICAST_MessageHeader *msg,
+                             enum GNUNET_PSYCSTORE_MessageFlags flags)
+{
+  struct OperationClosure *opcls = cls;
+  struct Channel *chn = opcls->chn;
+  client_send_mcast_msg (chn, msg, GNUNET_PSYC_MESSAGE_HISTORIC);
+  return GNUNET_YES;
+}
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_fragment_get() for multicast replay.
+ */
+static void
+store_recv_fragment_history_result (void *cls, int64_t result, const char *err_msg)
+{
+  struct OperationClosure *opcls = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p History replay #%" PRIu64 ": "
+              "PSYCSTORE returned %" PRId64 " (%s)\n",
+              opcls->chn, opcls->op_id, result, err_msg);
+
+  client_send_result (opcls->client, opcls->op_id, result, err_msg);
+}
+
+
 /**
  * Client requests channel history from PSYCstore.
  */
 static void
-client_recv_story_request (void *cls, struct GNUNET_SERVER_Client *client,
-                           const struct GNUNET_MessageHeader *msg)
+client_recv_history_replay (void *cls, struct GNUNET_SERVER_Client *client,
+                            const struct GNUNET_MessageHeader *msg)
+{
+  struct Channel *
+    chn = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  GNUNET_assert (NULL != chn);
+
+  const struct HistoryRequest *
+    req = (const struct HistoryRequest *) msg;
+
+  struct OperationClosure *opcls = GNUNET_malloc (sizeof (*opcls));
+  opcls->client = client;
+  opcls->chn = chn;
+  opcls->op_id = req->op_id;
+
+  if (0 == req->message_limit)
+    GNUNET_PSYCSTORE_message_get (store, &chn->pub_key, NULL,
+                                  GNUNET_ntohll (req->start_message_id),
+                                  GNUNET_ntohll (req->end_message_id),
+                                  &store_recv_fragment_history,
+                                  &store_recv_fragment_history_result, opcls);
+  else
+    GNUNET_PSYCSTORE_message_get_latest (store, &chn->pub_key, NULL,
+                                         GNUNET_ntohll (req->message_limit),
+                                         &store_recv_fragment_history,
+                                         &store_recv_fragment_history_result,
+                                         opcls);
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Received state var from PSYCstore, send it to client.
+ */
+static int
+store_recv_state_var (void *cls, const char *name,
+                      const void *value, size_t value_size)
+{
+  struct OperationClosure *opcls = cls;
+  struct OperationResult *op;
+
+  if (NULL != name)
+  {
+    uint16_t name_size = strnlen (name, GNUNET_PSYC_MODIFIER_MAX_PAYLOAD) + 1;
+    struct GNUNET_PSYC_MessageModifier *mod;
+    op = GNUNET_malloc (sizeof (*op) + sizeof (*mod) + name_size + value_size);
+    op->header.size = htons (sizeof (*op) + sizeof (*mod) + name_size + value_size);
+    op->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+    op->op_id = opcls->op_id;
+
+    mod = (struct GNUNET_PSYC_MessageModifier *) &op[1];
+    mod->header.size = htons (sizeof (*mod) + name_size + value_size);
+    mod->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER);
+    mod->name_size = htons (name_size);
+    mod->value_size = htonl (value_size);
+    mod->oper = htons (GNUNET_ENV_OP_ASSIGN);
+    memcpy (&mod[1], name, name_size);
+    memcpy (((char *) &mod[1]) + name_size, value, value_size);
+  }
+  else
+  {
+    struct GNUNET_MessageHeader *mod;
+    op = GNUNET_malloc (sizeof (*op) + sizeof (*mod) + value_size);
+    op->header.size = htons (sizeof (*op) + sizeof (*mod) + value_size);
+    op->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+    op->op_id = opcls->op_id;
+
+    mod = (struct GNUNET_MessageHeader *) &op[1];
+    mod->size = htons (sizeof (*mod) + value_size);
+    mod->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);
+    memcpy (&mod[1], value, value_size);
+  }
+
+  GNUNET_SERVER_notification_context_add (nc, opcls->client);
+  GNUNET_SERVER_notification_context_unicast (nc, opcls->client, &op->header,
+                                              GNUNET_NO);
+  return GNUNET_YES;
+}
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_state_get()
+ * or GNUNET_PSYCSTORE_state_get_prefix()
+ */
+static void
+store_recv_state_result (void *cls, int64_t result, const char *err_msg)
 {
+  struct OperationClosure *opcls = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p History replay #%" PRIu64 ": "
+              "PSYCSTORE returned %" PRId64 " (%s)\n",
+              opcls->chn, opcls->op_id, result, err_msg);
 
+  client_send_result (opcls->client, opcls->op_id, result, err_msg);
 }
 
 
@@ -2109,7 +2294,30 @@ static void
 client_recv_state_get (void *cls, struct GNUNET_SERVER_Client *client,
                        const struct GNUNET_MessageHeader *msg)
 {
+  struct Channel *
+    chn = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  GNUNET_assert (NULL != chn);
+
+  const struct StateRequest *
+    req = (const struct StateRequest *) msg;
 
+  uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
+  const char *name = (const char *) &req[1];
+  if (0 == name_size || '\0' != name[name_size - 1])
+  {
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  struct OperationClosure *opcls = GNUNET_malloc (sizeof (*opcls));
+  opcls->client = client;
+  opcls->chn = chn;
+  opcls->op_id = req->op_id;
+
+  GNUNET_PSYCSTORE_state_get (store, &chn->pub_key, name,
+                              &store_recv_state_var,
+                              &store_recv_state_result, opcls);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -2120,6 +2328,30 @@ static void
 client_recv_state_get_prefix (void *cls, struct GNUNET_SERVER_Client *client,
                               const struct GNUNET_MessageHeader *msg)
 {
+  struct Channel *
+    chn = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  GNUNET_assert (NULL != chn);
+
+  const struct StateRequest *
+    req = (const struct StateRequest *) msg;
+
+  uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
+  const char *name = (const char *) &req[1];
+  if (0 == name_size || '\0' != name[name_size - 1])
+  {
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  struct OperationClosure *opcls = GNUNET_malloc (sizeof (*opcls));
+  opcls->client = client;
+  opcls->chn = chn;
+  opcls->op_id = req->op_id;
+
+  GNUNET_PSYCSTORE_state_get_prefix (store, &chn->pub_key, name,
+                                     &store_recv_state_var,
+                                     &store_recv_state_result, opcls);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 
 }
 
@@ -2140,8 +2372,8 @@ static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
   { &client_recv_membership_store, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE, 0 },
 
-  { &client_recv_story_request, NULL,
-    GNUNET_MESSAGE_TYPE_PSYC_STORY_REQUEST, 0 },
+  { &client_recv_history_replay, NULL,
+    GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY, 0 },
 
   { &client_recv_state_get, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_STATE_GET, 0 },
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 21131e7d3..f6d40ddb4 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -112,30 +112,39 @@ struct ChannelMembershipStoreRequest
    */
   struct GNUNET_MessageHeader header;
 
-  uint32_t reserved;
+  uint32_t reserved GNUNET_PACKED;
+
+  uint64_t op_id GNUNET_PACKED;
 
   struct GNUNET_CRYPTO_EcdsaPublicKey slave_key;
 
-  uint64_t announced_at;
+  uint64_t announced_at GNUNET_PACKED;
 
-  uint64_t effective_since;
+  uint64_t effective_since GNUNET_PACKED;
 
   uint8_t did_join;
 };
 
 
-struct StoryRequest
+struct HistoryRequest
 {
   /**
-   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STORY_REQUEST
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_HISTORY_REQUEST
    */
   struct GNUNET_MessageHeader header;
 
-  uint64_t op_id;
+  uint32_t reserved GNUNET_PACKED;
+
+  /**
+   * ID for this operation.
+   */
+  uint64_t op_id GNUNET_PACKED;
+
+  uint64_t start_message_id GNUNET_PACKED;
 
-  uint64_t start_message_id;
+  uint64_t end_message_id GNUNET_PACKED;
 
-  uint64_t end_message_id;
+  uint64_t message_limit GNUNET_PACKED;
 };
 
 
@@ -148,10 +157,12 @@ struct StateRequest
    */
   struct GNUNET_MessageHeader header;
 
+  uint32_t reserved GNUNET_PACKED;
+
   /**
    * ID for this operation.
    */
-  uint64_t op_id;
+  uint64_t op_id GNUNET_PACKED;
 
   /* Followed by NUL-terminated name. */
 };
@@ -160,25 +171,6 @@ struct StateRequest
 /**** service -> library ****/
 
 
-struct CountersResult
-{
-  /**
-   * Type: GNUNET_MESSAGE_TYPE_PSYC_RESULT_COUNTERS
-   */
-  struct GNUNET_MessageHeader header;
-
-  /**
-   * Status code for the operation.
-   */
-  int32_t result_code GNUNET_PACKED;
-
-  /**
-   * Last message ID sent to the channel.
-   */
-  uint64_t max_message_id;
-};
-
-
 /**
  * Answer from service to client about last operation.
  */
@@ -192,23 +184,22 @@ struct OperationResult
    */
   struct GNUNET_MessageHeader header;
 
+  uint32_t reserved GNUNET_PACKED;
+
   /**
    * Operation ID.
    */
-  uint32_t op_id GNUNET_PACKED;
+  uint64_t op_id GNUNET_PACKED;
 
   /**
    * Status code for the operation.
    */
-  int64_t result_code GNUNET_PACKED;
+  uint64_t result_code GNUNET_PACKED;
 
   /* Followed by:
    * - on error: NUL-terminated error message
    * - on success: one of the following message types
    *
-   *   For a STORY_RESULT:
-   *   - GNUNET_MESSAGE_TYPE_PSYC_MESSAGE
-   *
    *   For a STATE_RESULT, one of:
    *   - GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER
    *   - GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index ca25b1b01..8cce89704 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -43,6 +43,33 @@
 #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
 
 
+struct OperationListItem
+{
+  struct OperationListItem *prev;
+  struct OperationListItem *next;
+
+  /**
+   * Operation ID.
+   */
+  uint64_t op_id;
+
+  /**
+   * Continuation to invoke with the result of an operation.
+   */
+  GNUNET_PSYC_ResultCallback result_cb;
+
+  /**
+   * State variable result callback.
+   */
+  GNUNET_PSYC_StateVarCallback state_var_cb;
+
+  /**
+   * Closure for the callbacks.
+   */
+  void *cls;
+};
+
+
 /**
  * Handle to access PSYC channel operations for both the master and slaves.
  */
@@ -84,6 +111,21 @@ struct GNUNET_PSYC_Channel
   void *disconnect_cls;
 
   /**
+   * First operation in the linked list.
+   */
+  struct OperationListItem *op_head;
+
+  /**
+   * Last operation in the linked list.
+   */
+  struct OperationListItem *op_tail;
+
+  /**
+   * Last operation ID used.
+   */
+  uint64_t last_op_id;
+
+  /**
    * Are we polling for incoming messages right now?
    */
   uint8_t in_receive;
@@ -163,21 +205,82 @@ struct GNUNET_PSYC_SlaveTransmitHandle
 
 
 /**
- * Handle to a story telling operation.
+ * Get a fresh operation ID to distinguish between PSYCstore requests.
+ *
+ * @param h Handle to the PSYCstore service.
+ * @return next operation id to use
  */
-struct GNUNET_PSYC_Story
+static uint64_t
+op_get_next_id (struct GNUNET_PSYC_Channel *chn)
 {
-
-};
+  return ++chn->last_op_id;
+}
 
 
 /**
- * Handle for a state query operation.
+ * Find operation by ID.
+ *
+ * @return Operation, or NULL if none found.
  */
-struct GNUNET_PSYC_StateQuery
+static struct OperationListItem *
+op_find_by_id (struct GNUNET_PSYC_Channel *chn, uint64_t op_id)
 {
+  struct OperationListItem *op = chn->op_head;
+  while (NULL != op)
+  {
+    if (op->op_id == op_id)
+      return op;
+    op = op->next;
+  }
+  return NULL;
+}
 
-};
+
+static uint64_t
+op_add (struct GNUNET_PSYC_Channel *chn, GNUNET_PSYC_ResultCallback result_cb,
+        void *cls)
+{
+  if (NULL == result_cb)
+    return 0;
+
+  struct OperationListItem *op = GNUNET_malloc (sizeof (*op));
+  op->op_id = op_get_next_id (chn);
+  op->result_cb = result_cb;
+  op->cls = cls;
+  GNUNET_CONTAINER_DLL_insert_tail (chn->op_head, chn->op_tail, op);
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%p Added operation #%" PRIu64 "\n", chn, op->op_id);
+  return op->op_id;
+}
+
+
+static int
+op_result (struct GNUNET_PSYC_Channel *chn, uint64_t op_id,
+           int64_t result_code, const char *err_msg)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%p Received result for operation #%" PRIu64 ": %" PRId64 " (%s)\n",
+       chn, op_id, result_code, err_msg);
+  if (0 == op_id)
+    return GNUNET_NO;
+
+  struct OperationListItem *op = op_find_by_id (chn, op_id);
+  if (NULL == op)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Could not find operation #%" PRIu64 "\n", op_id);
+    return GNUNET_NO;
+  }
+
+  GNUNET_CONTAINER_DLL_remove (chn->op_head, chn->op_tail, op);
+
+  if (NULL != op->result_cb)
+    op->result_cb (op->cls, result_code, err_msg);
+
+  GNUNET_free (op);
+  return GNUNET_YES;
+}
 
 
 static void
@@ -203,6 +306,79 @@ channel_recv_disconnect (void *cls,
 
 
 static void
+channel_recv_result (void *cls,
+                     struct GNUNET_CLIENT_MANAGER_Connection *client,
+                     const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PSYC_Channel *
+    chn = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*chn));
+
+  uint16_t size = ntohs (msg->size);
+  const struct OperationResult *res = (const struct OperationResult *) msg;
+  const char *err_msg = NULL;
+
+  if (sizeof (struct OperationResult) < size)
+  {
+    err_msg = (const char *) &res[1];
+    if ('\0' != err_msg[size - sizeof (struct OperationResult) - 1])
+    {
+      GNUNET_break (0);
+      err_msg = NULL;
+    }
+  }
+
+  op_result (chn, GNUNET_ntohll (res->op_id),
+             GNUNET_ntohll (res->result_code) + INT64_MIN, err_msg);
+}
+
+
+static void
+channel_recv_state_result (void *cls,
+                           struct GNUNET_CLIENT_MANAGER_Connection *client,
+                           const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PSYC_Channel *
+    chn = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*chn));
+
+  const struct OperationResult *res = (const struct OperationResult *) msg;
+  struct OperationListItem *op = op_find_by_id (chn, GNUNET_ntohll (res->op_id));
+  if (NULL == op || NULL == op->state_var_cb)
+    return;
+
+  const struct GNUNET_MessageHeader *modc = (struct GNUNET_MessageHeader *) &op[1];
+  uint16_t modc_size = ntohs (modc->size);
+  if (ntohs (msg->size) - sizeof (*msg) != modc_size)
+  {
+    GNUNET_break (0);
+    return;
+  }
+  switch (ntohs (modc->type))
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+  {
+    const struct GNUNET_PSYC_MessageModifier *
+      mod = (const struct GNUNET_PSYC_MessageModifier *) modc;
+
+    const char *name = (const char *) &mod[1];
+    uint16_t name_size = ntohs (mod->name_size);
+    if ('\0' != name[name_size - 1])
+    {
+      GNUNET_break (0);
+      return;
+    }
+    op->state_var_cb (op->cls, name, name + name_size, ntohs (mod->value_size));
+    break;
+  }
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+    op->state_var_cb (op->cls, NULL, (const char *) &modc[1],
+                      modc_size - sizeof (*modc));
+    break;
+  }
+}
+
+
+static void
 channel_recv_message (void *cls,
                       struct GNUNET_CLIENT_MANAGER_Connection *client,
                       const struct GNUNET_MessageHeader *msg)
@@ -234,9 +410,16 @@ master_recv_start_ack (void *cls,
     mst = GNUNET_CLIENT_MANAGER_get_user_context_ (client,
                                                    sizeof (struct GNUNET_PSYC_Channel));
 
-  struct CountersResult *cres = (struct CountersResult *) msg;
+  struct GNUNET_PSYC_CountersResultMessage *
+    cres = (struct GNUNET_PSYC_CountersResultMessage *) msg;
+  int32_t result = ntohl (cres->result_code) + INT32_MIN;
+  if (GNUNET_OK != result && GNUNET_NO != result)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master.\n");
+    GNUNET_break (0);
+  }
   if (NULL != mst->start_cb)
-    mst->start_cb (mst->cb_cls, GNUNET_ntohll (cres->max_message_id));
+    mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
 }
 
 
@@ -279,9 +462,16 @@ slave_recv_join_ack (void *cls,
   struct GNUNET_PSYC_Slave *
     slv = GNUNET_CLIENT_MANAGER_get_user_context_ (client,
                                                    sizeof (struct GNUNET_PSYC_Channel));
-  struct CountersResult *cres = (struct CountersResult *) msg;
+  struct GNUNET_PSYC_CountersResultMessage *
+    cres = (struct GNUNET_PSYC_CountersResultMessage *) msg;
+  int32_t result = ntohl (cres->result_code) + INT32_MIN;
+  if (GNUNET_YES != result && GNUNET_NO != result)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n");
+    GNUNET_break (0);
+  }
   if (NULL != slv->connect_cb)
-    slv->connect_cb (slv->cb_cls, GNUNET_ntohll (cres->max_message_id));
+    slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
 }
 
 
@@ -317,12 +507,20 @@ static struct GNUNET_CLIENT_MANAGER_MessageHandler master_handlers[] =
 
   { &master_recv_start_ack, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK,
-    sizeof (struct CountersResult), GNUNET_NO },
+    sizeof (struct GNUNET_PSYC_CountersResultMessage), GNUNET_NO },
 
   { &master_recv_join_request, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
     sizeof (struct GNUNET_PSYC_JoinRequestMessage), GNUNET_YES },
 
+  { &channel_recv_state_result, NULL,
+    GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
+    sizeof (struct OperationResult), GNUNET_YES },
+
+  { &channel_recv_result, NULL,
+    GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
+    sizeof (struct OperationResult), GNUNET_YES },
+
   { &channel_recv_disconnect, NULL, 0, 0, GNUNET_NO },
 
   { NULL, NULL, 0, 0, GNUNET_NO }
@@ -341,12 +539,20 @@ static struct GNUNET_CLIENT_MANAGER_MessageHandler slave_handlers[] =
 
   { &slave_recv_join_ack, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK,
-    sizeof (struct CountersResult), GNUNET_NO },
+    sizeof (struct GNUNET_PSYC_CountersResultMessage), GNUNET_NO },
 
   { &slave_recv_join_decision, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
     sizeof (struct GNUNET_PSYC_JoinDecisionMessage), GNUNET_YES },
 
+  { &channel_recv_state_result, NULL,
+    GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
+    sizeof (struct OperationResult), GNUNET_YES },
+
+  { &channel_recv_result, NULL,
+    GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
+    sizeof (struct OperationResult), GNUNET_YES },
+
   { &channel_recv_disconnect, NULL, 0, 0, GNUNET_NO },
 
   { NULL, NULL, 0, 0, GNUNET_NO }
@@ -808,7 +1014,9 @@ void
 GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
                                const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
                                uint64_t announced_at,
-                               uint64_t effective_since)
+                               uint64_t effective_since,
+                               GNUNET_PSYC_ResultCallback result_cb,
+                               void *cls)
 {
   struct ChannelMembershipStoreRequest *req = GNUNET_malloc (sizeof (*req));
   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
@@ -817,6 +1025,8 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
   req->announced_at = GNUNET_htonll (announced_at);
   req->effective_since = GNUNET_htonll (effective_since);
   req->did_join = GNUNET_YES;
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
+
   GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
@@ -845,7 +1055,9 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
 void
 GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn,
                                   const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
-                                  uint64_t announced_at)
+                                  uint64_t announced_at,
+                                  GNUNET_PSYC_ResultCallback result_cb,
+                                  void *cls)
 {
   struct ChannelMembershipStoreRequest *req = GNUNET_malloc (sizeof (*req));
   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
@@ -853,57 +1065,85 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn,
   req->slave_key = *slave_key;
   req->announced_at = GNUNET_htonll (announced_at);
   req->did_join = GNUNET_NO;
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
+
   GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
 
 /**
- * Request to be told the message history of the channel.
+ * Request to replay a part of the message history of the channel.
  *
  * Historic messages (but NOT the state at the time) will be replayed (given to
  * the normal method handlers) if available and if access is permitted.
  *
- * To get the latest message, use 0 for both the start and end message ID.
- *
- * @param channel Which channel should be replayed?
- * @param start_message_id Earliest interesting point in history.
- * @param end_message_id Last (exclusive) interesting point in history.
- * @param message_cb Function to invoke on message parts received from the story.
- * @param finish_cb Function to call when the requested story has been fully
- *        told (counting message IDs might not suffice, as some messages
- *        might be secret and thus the listener would not know the story is
- *        finished without being told explicitly) once this function
- *        has been called, the client must not call
- *        GNUNET_PSYC_channel_story_tell_cancel() anymore.
- * @param cls Closure for the callbacks.
- *
- * @return Handle to cancel story telling operation.
+ * @param channel
+ *        Which channel should be replayed?
+ * @param start_message_id
+ *        Earliest interesting point in history.
+ * @param end_message_id
+ *        Last (inclusive) interesting point in history.
+ * FIXME: @param method_prefix
+ *        Retrieve only messages with a matching method prefix.
+ * @param result_cb
+ *        Function to call when the requested history has been fully replayed.
+ * @param cls
+ *        Closure for the callbacks.
+ *
+ * @return Handle to cancel history replay operation.
  */
-struct GNUNET_PSYC_Story *
-GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel,
-                                uint64_t start_message_id,
-                                uint64_t end_message_id,
-                                GNUNET_PSYC_MessageCallback message_cb,
-                                GNUNET_PSYC_MessagePartCallback message_part_cb,
-                                GNUNET_PSYC_FinishCallback finish_cb,
-                                void *cls)
+void
+GNUNET_PSYC_channel_history_replay (struct GNUNET_PSYC_Channel *chn,
+                                    uint64_t start_message_id,
+                                    uint64_t end_message_id,
+                                    /* FIXME: const char *method_prefix, */
+                                    GNUNET_PSYC_ResultCallback result_cb,
+                                    void *cls)
 {
-  return NULL;
+  struct HistoryRequest *req = GNUNET_malloc (sizeof (*req));
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY);
+  req->header.size = htons (sizeof (*req));
+  req->start_message_id = GNUNET_htonll (start_message_id);
+  req->end_message_id = GNUNET_htonll (end_message_id);
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
+
+  GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
 
 /**
- * Abort story telling.
+ * Request to replay the latest messages from the message history of the channel.
  *
- * This function must not be called from within method handlers (as given to
- * GNUNET_PSYC_slave_join()) of the slave.
+ * Historic messages (but NOT the state at the time) will be replayed (given to
+ * the normal method handlers) if available and if access is permitted.
  *
- * @param story Story telling operation to stop.
+ * @param channel
+ *        Which channel should be replayed?
+ * @param message_limit
+ *        Maximum number of messages to replay.
+ * FIXME: @param method_prefix
+ *        Retrieve only messages with a matching method prefix.
+ * @param result_cb
+ *        Function to call when the requested history has been fully replayed.
+ * @param cls
+ *        Closure for the callbacks.
+ *
+ * @return Handle to cancel history replay operation.
  */
 void
-GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story)
+GNUNET_PSYC_channel_history_replay_latest (struct GNUNET_PSYC_Channel *chn,
+                                           uint64_t message_limit,
+                                           /* FIXME: const char *method_prefix, */
+                                           GNUNET_PSYC_ResultCallback result_cb,
+                                           void *cls)
 {
+  struct HistoryRequest *req = GNUNET_malloc (sizeof (*req));
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY);
+  req->header.size = htons (sizeof (*req));
+  req->message_limit = GNUNET_htonll (message_limit);
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
 
+  GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
 
@@ -914,22 +1154,35 @@ GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story)
  * less-specific name is matched; for example, requesting "_a_b" will match "_a"
  * if "_a_b" does not exist.
  *
- * @param channel Channel handle.
- * @param full_name Full name of the requested variable, the actual variable
- *        returned might have a shorter name..
- * @param cb Function called once when a matching state variable is found.
+ * @param channel
+ *        Channel handle.
+ * @param full_name
+ *        Full name of the requested variable.
+ *        The actual variable returned might have a shorter name.
+ * @param var_cb
+ *        Function called once when a matching state variable is found.
  *        Not called if there's no matching state variable.
- * @param cb_cls Closure for the callbacks.
- *
- * @return Handle that can be used to cancel the query operation.
+ * @param result_cb
+ *        Function called after the operation finished.
+ *        (i.e. all state variables have been returned via @a state_cb)
+ * @param cls
+ *        Closure for the callbacks.
  */
-struct GNUNET_PSYC_StateQuery *
-GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel,
+void
+GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *chn,
                                const char *full_name,
-                               GNUNET_PSYC_StateCallback cb,
-                               void *cb_cls)
+                               GNUNET_PSYC_StateVarCallback var_cb,
+                               GNUNET_PSYC_ResultCallback result_cb,
+                               void *cls)
 {
-  return NULL;
+  size_t name_size = strlen (full_name) + 1;
+  struct StateRequest *req = GNUNET_malloc (sizeof (*req) + name_size);
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_GET);
+  req->header.size = htons (sizeof (*req) + name_size);
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
+  memcpy (&req[1], full_name, name_size);
+
+  GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
 
@@ -943,33 +1196,34 @@ GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel,
  * The @a state_cb is invoked on all matching state variables asynchronously, as
  * the state is stored in and retrieved from the PSYCstore,
  *
- * @param channel Channel handle.
- * @param name_prefix Prefix of the state variable name to match.
- * @param cb Function to call with the matching state variables.
- * @param cb_cls Closure for the callbacks.
- *
- * @return Handle that can be used to cancel the query operation.
- */
-struct GNUNET_PSYC_StateQuery *
-GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *channel,
-                                      const char *name_prefix,
-                                      GNUNET_PSYC_StateCallback cb,
-                                      void *cb_cls)
-{
-  return NULL;
-}
-
-
-/**
- * Cancel a state query operation.
- *
- * @param query Handle for the operation to cancel.
+ * @param channel
+ *        Channel handle.
+ * @param name_prefix
+ *        Prefix of the state variable name to match.
+ * @param var_cb
+ *        Function called once when a matching state variable is found.
+ *        Not called if there's no matching state variable.
+ * @param result_cb
+ *        Function called after the operation finished.
+ *        (i.e. all state variables have been returned via @a state_cb)
+ * @param cls
+ *        Closure for the callbacks.
  */
 void
-GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateQuery *query)
+GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *chn,
+                                      const char *name_prefix,
+                                      GNUNET_PSYC_StateVarCallback var_cb,
+                                      GNUNET_PSYC_ResultCallback result_cb,
+                                      void *cls)
 {
+  size_t name_size = strlen (name_prefix) + 1;
+  struct StateRequest *req = GNUNET_malloc (sizeof (*req) + name_size);
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_GET);
+  req->header.size = htons (sizeof (*req) + name_size);
+  req->op_id = GNUNET_htonll (op_add (chn, result_cb, cls));
+  memcpy (&req[1], name_prefix, name_size);
 
+  GNUNET_CLIENT_MANAGER_transmit (chn->client, &req->header);
 }
 
-
 /* end of psyc_api.c */
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 5eadef62c..044895809 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -35,6 +35,7 @@
 #include "gnunet_env_lib.h"
 #include "gnunet_psyc_util_lib.h"
 #include "gnunet_psyc_service.h"
+#include "gnunet_core_service.h"
 
 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
 
@@ -45,6 +46,9 @@ int res;
 
 const struct GNUNET_CONFIGURATION_Handle *cfg;
 
+struct GNUNET_CORE_Handle *core;
+struct GNUNET_PeerIdentity this_peer;
+
 /**
  * Handle for task for timeout termination.
  */
@@ -53,6 +57,8 @@ GNUNET_SCHEDULER_TaskIdentifier end_badly_task;
 struct GNUNET_PSYC_Master *mst;
 struct GNUNET_PSYC_Slave *slv;
 
+struct GNUNET_PSYC_Channel *mst_chn, *slv_chn;
+
 struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key;
 struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key;
 
@@ -80,9 +86,19 @@ uint8_t join_req_count;
 
 enum
 {
-  TEST_NONE,
-  TEST_SLAVE_TRANSMIT,
-  TEST_MASTER_TRANSMIT,
+  TEST_NONE = 0,
+  TEST_MASTER_START = 1,
+  TEST_SLAVE_JOIN = 2,
+  TEST_SLAVE_TRANSMIT = 3,
+  TEST_MASTER_TRANSMIT = 4,
+  TEST_MASTER_HISTORY_REPLAY_LATEST = 5,
+  TEST_SLAVE_HISTORY_REPLAY_LATEST = 6,
+  TEST_MASTER_HISTORY_REPLAY = 7,
+  TEST_SLAVE_HISTORY_REPLAY = 8,
+  TEST_MASTER_STATE_GET = 9,
+  TEST_SLAVE_STATE_GET = 10,
+  TEST_MASTER_STATE_GET_PREFIX = 11,
+  TEST_SLAVE_STATE_GET_PREFIX = 12,
 } test;
 
 
@@ -118,6 +134,11 @@ void slave_parted (void *cls)
 void
 cleanup ()
 {
+  if (NULL != core)
+  {
+    GNUNET_CORE_disconnect (core);
+    core = NULL;
+  }
   if (NULL != slv)
   {
     GNUNET_PSYC_slave_part (slv, GNUNET_NO, &slave_parted, NULL);
@@ -177,13 +198,203 @@ end ()
 
 
 void
+state_get_var (void *cls, const char *name, const void *value, size_t value_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Got state var: %s\n%.*s\n", name, value_size, value);  
+}
+
+
+/*** Slave state_get_prefix() ***/
+
+void
+slave_state_get_prefix_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_state_get_prefix:\t%" PRId64 " (%s)\n", result, err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  end ();
+}
+
+
+void
+slave_state_get_prefix ()
+{
+  test = TEST_SLAVE_STATE_GET_PREFIX;
+  GNUNET_PSYC_channel_state_get_prefix (slv_chn, "_foo", &state_get_var,
+                                        &slave_state_get_prefix_result, NULL);
+}
+
+
+/*** Master state_get_prefix() ***/
+
+
+void
+master_state_get_prefix_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "master_state_get_prefix:\t%" PRId64 " (%s)\n", result, err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  slave_state_get_prefix ();
+}
+
+
+void
+master_state_get_prefix ()
+{
+  test = TEST_MASTER_STATE_GET_PREFIX;
+  GNUNET_PSYC_channel_state_get_prefix (mst_chn, "_foo", &state_get_var,
+                                        &master_state_get_prefix_result, NULL);
+}
+
+
+/*** Slave state_get() ***/
+
+
+void
+slave_state_get_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_state_get:\t%" PRId64 " (%s)\n", result, err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  master_state_get_prefix ();
+}
+
+
+void
+slave_state_get ()
+{
+  test = TEST_SLAVE_STATE_GET;
+  GNUNET_PSYC_channel_state_get (slv_chn, "_foo_bar_baz", &state_get_var,
+                                 &slave_state_get_result, NULL);
+}
+
+
+/*** Master state_get() ***/
+
+
+void
+master_state_get_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "master_state_get:\t%" PRId64 " (%s)\n", result, err_msg);
+  // FIXME: GNUNET_assert (1 == result);
+  slave_state_get ();
+}
+
+
+void
+master_state_get ()
+{
+  test = TEST_MASTER_STATE_GET;
+  GNUNET_PSYC_channel_state_get (mst_chn, "_foo_bar_baz", &state_get_var,
+                                 &master_state_get_result, NULL);
+}
+
+
+/*** Slave history_replay() ***/
+
+void
+slave_history_replay_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_history_replay:\t%" PRId64 " (%s)\n", result, err_msg);
+  GNUNET_assert (9 == result);
+
+  master_state_get ();
+}
+
+
+void
+slave_history_replay ()
+{
+  test = TEST_SLAVE_HISTORY_REPLAY;
+  GNUNET_PSYC_channel_history_replay (slv_chn, 1, 1,
+                                      &slave_history_replay_result,
+                                      NULL);
+}
+
+
+/*** Master history_replay() ***/
+
+
+void
+master_history_replay_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "master_history_replay:\t%" PRId64 " (%s)\n", result, err_msg);
+  GNUNET_assert (9 == result);
+
+  slave_history_replay ();
+}
+
+
+void
+master_history_replay ()
+{
+  test = TEST_MASTER_HISTORY_REPLAY;
+  GNUNET_PSYC_channel_history_replay (mst_chn, 1, 1,
+                                      &master_history_replay_result,
+                                      NULL);
+}
+
+
+/*** Slave history_replay_latest() ***/
+
+
+void
+slave_history_replay_latest_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_history_replay_latest:\t%" PRId64 " (%s)\n", result, err_msg);
+  GNUNET_assert (9 == result);
+
+  master_history_replay ();
+}
+
+
+void
+slave_history_replay_latest ()
+{
+  test = TEST_SLAVE_HISTORY_REPLAY_LATEST;
+  GNUNET_PSYC_channel_history_replay_latest (slv_chn, 1,
+                                             &slave_history_replay_latest_result,
+                                             NULL);
+}
+
+
+/*** Master history_replay_latest() ***/
+
+
+void
+master_history_replay_latest_result (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "master_history_replay_latest:\t%" PRId64 " (%s)\n", result, err_msg);
+  GNUNET_assert (9 == result);
+
+  slave_history_replay_latest ();
+}
+
+
+void
+master_history_replay_latest ()
+{
+  test = TEST_MASTER_HISTORY_REPLAY_LATEST;
+  GNUNET_PSYC_channel_history_replay_latest (mst_chn, 1,
+                                             &master_history_replay_latest_result,
+                                             NULL);
+}
+
+
+void
 master_message_cb (void *cls, uint64_t message_id, uint32_t flags,
                    const struct GNUNET_PSYC_MessageHeader *msg)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Master got PSYC message fragment of size %u "
-              "belonging to message ID %llu with flags %x\n",
-              ntohs (msg->header.size), message_id, flags);
+              "Test #%d: Master got PSYC message fragment of size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, ntohs (msg->header.size), message_id, flags);
   // FIXME
 }
 
@@ -196,7 +407,7 @@ master_message_part_cb (void *cls, uint64_t message_id,
   if (NULL == msg)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Error while receiving message %llu\n", message_id);
+                "Error while receiving message %" PRIu64 "\n", message_id);
     return;
   }
 
@@ -204,9 +415,9 @@ master_message_part_cb (void *cls, uint64_t message_id,
   uint16_t size = ntohs (msg->size);
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Master got message part of type %u and size %u "
-              "belonging to message ID %llu with flags %x\n",
-              type, size, message_id, flags);
+              "Test #%d: Master got message part of type %u and size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, type, size, message_id, flags);
 
   switch (test)
   {
@@ -227,6 +438,18 @@ master_message_part_cb (void *cls, uint64_t message_id,
   case TEST_MASTER_TRANSMIT:
     break;
 
+  case TEST_MASTER_HISTORY_REPLAY:
+  case TEST_MASTER_HISTORY_REPLAY_LATEST:
+    if (GNUNET_PSYC_MESSAGE_HISTORIC != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Test #%d: Unexpected flags for historic message: %x" PRIu32 "\n",
+                  flags);
+      GNUNET_assert (0);
+      return;
+    }
+    break;
+
   default:
     GNUNET_assert (0);
   }
@@ -238,9 +461,9 @@ slave_message_cb (void *cls, uint64_t message_id, uint32_t flags,
                   const struct GNUNET_PSYC_MessageHeader *msg)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Slave got PSYC message fragment of size %u "
-              "belonging to message ID %llu with flags %x\n",
-              ntohs (msg->header.size), message_id, flags);
+              "Test #%d: Slave got PSYC message fragment of size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, ntohs (msg->header.size), message_id, flags);
   // FIXME
 }
 
@@ -253,7 +476,7 @@ slave_message_part_cb (void *cls, uint64_t message_id,
   if (NULL == msg)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Error while receiving message %llu\n", message_id);
+                "Error while receiving message " PRIu64 "\n", message_id);
     return;
   }
 
@@ -261,15 +484,27 @@ slave_message_part_cb (void *cls, uint64_t message_id,
   uint16_t size = ntohs (msg->size);
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Slave got message part of type %u and size %u "
-              "belonging to message ID %llu with flags %x\n",
-              type, size, message_id, flags);
+              "Test #%d: Slave got message part of type %u and size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, type, size, message_id, flags);
 
   switch (test)
   {
   case TEST_MASTER_TRANSMIT:
     if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type)
-      end ();
+      master_history_replay_latest ();
+    break;
+
+  case TEST_SLAVE_HISTORY_REPLAY:
+  case TEST_SLAVE_HISTORY_REPLAY_LATEST:
+    if (GNUNET_PSYC_MESSAGE_HISTORIC != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Test #%d: Unexpected flags for historic message: %x" PRIu32 "\n",
+                  flags);
+      GNUNET_assert (0);
+      return;
+    }
     break;
 
   default:
@@ -417,7 +652,6 @@ slave_transmit ()
 {
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave sending request to master.\n");
-
   test = TEST_SLAVE_TRANSMIT;
 
   tmit = GNUNET_new (struct TransmitClosure);
@@ -438,6 +672,29 @@ slave_transmit ()
 
 
 void
+slave_remove_cb (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_remove:\t%" PRId64 " (%s)\n", result, err_msg);
+
+  slave_transmit ();
+}
+
+
+void
+slave_add_cb (void *cls, int64_t result, const char *err_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "slave_add:\t%" PRId64 " (%s)\n", result, err_msg);
+
+  struct GNUNET_PSYC_Channel *chn = cls;
+  GNUNET_PSYC_channel_slave_remove (chn, &slave_pub_key, 2,
+                                    &slave_remove_cb, chn);
+
+}
+
+
+void
 join_decision_cb (void *cls,
                   const struct GNUNET_PSYC_JoinDecisionMessage *dcsn,
                   int is_admitted,
@@ -453,7 +710,8 @@ join_decision_cb (void *cls,
     return;
   }
 
-  slave_transmit ();
+  struct GNUNET_PSYC_Channel *chn = GNUNET_PSYC_master_get_channel (mst);
+  GNUNET_PSYC_channel_slave_add (chn, &slave_pub_key, 2, 2, &slave_add_cb, chn);
 }
 
 
@@ -473,19 +731,17 @@ join_request_cb (void *cls,
   /* Reject first request */
   int is_admitted = (0 < join_req_count++) ? GNUNET_YES : GNUNET_NO;
   GNUNET_PSYC_join_decision (jh, is_admitted, 0, NULL, NULL);
-
-  /* Membership store */
-  struct GNUNET_PSYC_Channel *chn = GNUNET_PSYC_master_get_channel (mst);
-  GNUNET_PSYC_channel_slave_add (chn, slave_key, 2, 2);
-  GNUNET_PSYC_channel_slave_remove (chn, &slave_pub_key, 2);
 }
 
 
 void
-slave_connect_cb (void *cls, uint64_t max_message_id)
+slave_connect_cb (void *cls, int result, uint64_t max_message_id)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Slave connected: %lu\n", max_message_id);
+              "Slave connected: %d, max_message_id: %" PRIu64 "\n",
+              result, max_message_id);
+  GNUNET_assert (TEST_SLAVE_JOIN == test);
+  GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
 }
 
 
@@ -493,8 +749,9 @@ void
 slave_join ()
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Joining slave.\n");
+  test = TEST_SLAVE_JOIN;
 
-  struct GNUNET_PeerIdentity origin = {}; // FIXME: this peer
+  struct GNUNET_PeerIdentity origin = this_peer;
   struct GNUNET_ENV_Environment *env = GNUNET_ENV_environment_create ();
   GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
                               "_foo", "bar baz", 7);
@@ -507,6 +764,7 @@ slave_join ()
                                 &slave_message_cb, &slave_message_part_cb,
                                 &slave_connect_cb, &join_decision_cb, NULL,
                                 join_msg);
+  slv_chn = GNUNET_PSYC_slave_get_channel (slv);
   GNUNET_ENV_environment_destroy (env);
 }
 
@@ -564,10 +822,13 @@ master_transmit ()
 
 
 void
-master_start_cb (void *cls, uint64_t max_message_id)
+master_start_cb (void *cls, int result, uint64_t max_message_id)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Master started: %" PRIu64 "\n", max_message_id);
+              "Master started: %d, max_message_id: %" PRIu64 "\n",
+              result, max_message_id);
+  GNUNET_assert (TEST_MASTER_START == test);
+  GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
   slave_join ();
 }
 
@@ -576,10 +837,12 @@ void
 master_start ()
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Starting master.\n");
+  test = TEST_MASTER_START;
   mst = GNUNET_PSYC_master_start (cfg, channel_key, GNUNET_PSYC_CHANNEL_PRIVATE,
                                   &master_start_cb, &join_request_cb,
                                   &master_message_cb, &master_message_part_cb,
                                   NULL);
+  mst_chn = GNUNET_PSYC_master_get_channel (mst);
 }
 
 void
@@ -589,6 +852,21 @@ schedule_master_start (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 }
 
 
+void
+core_connected (void *cls, const struct GNUNET_PeerIdentity *my_identity)
+{
+  this_peer = *my_identity;
+
+#if DEBUG_TEST_PSYC
+  master_start ();
+#else
+  /* Allow some time for the services to initialize. */
+  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+                                &schedule_master_start, NULL);
+#endif
+
+}
+
 /**
  * Main function of the test, run from scheduler.
  *
@@ -615,14 +893,8 @@ run (void *cls,
   GNUNET_CRYPTO_eddsa_key_get_public (channel_key, &channel_pub_key);
   GNUNET_CRYPTO_ecdsa_key_get_public (slave_key, &slave_pub_key);
 
-#if DEBUG_TEST_PSYC
-  master_start ();
-#else
-  /* Allow some time for the services to initialize. */
-  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
-                                &schedule_master_start, NULL);
-#endif
-  return;
+  core = GNUNET_CORE_connect (cfg, NULL, &core_connected, NULL, NULL,
+                              NULL, GNUNET_NO, NULL, GNUNET_NO, NULL);
 }