1 files changed, 53 insertions, 4 deletions

diff --git a/src/ipcpd/unicast/dir/dht.c b/src/ipcpd/unicast/dir/dht.c
index 8eeea800..c2cd33aa 100644
--- a/src/ipcpd/unicast/dir/dht.c
+++ b/src/ipcpd/unicast/dir/dht.c
@@ -1597,6 +1597,7 @@ static ssize_t dht_kv_get_contacts(const uint8_t *       key,
  fail_contact:
         while (i-- > 0)
                 dht_contact_msg__free_unpacked((*msgs)[i], NULL);
+
         free(*msgs);
         *msgs = NULL;
  fail_msgs:
@@ -1763,6 +1764,7 @@ static int split_bucket(struct bucket * b)
  fail_child:
         while (i-- > 0)
                 bucket_destroy(b->children[i]);
+
         return -1;
 }
 
@@ -2236,7 +2238,7 @@ static int dht_send_msg(dht_msg_t * msg,
 
         dht_msg__pack(msg, ssm_pk_buff_head(spb));
 
-        if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, spb) < 0) {
+        if (dt_write_packet(addr, QOS_CUBE_BE, dht.eid, spb, NULL) < 0) {
                 log_warn("%s write failed", DHT_CODE(msg));
                 goto fail_send;
         }
@@ -2849,6 +2851,7 @@ static dht_msg_t * do_dht_kv_find_node_req(const dht_find_req_msg_t * req)
  fail_msg:
         while (len-- > 0)
                 dht_contact_msg__free_unpacked(contacts[len], NULL);
+
         free(contacts);
  fail_contacts:
         return NULL;
@@ -2951,8 +2954,9 @@ static dht_msg_t * do_dht_kv_find_value_req(const dht_find_req_msg_t * req)
  fail_msg:
         freebufs(vals, n_vals);
  fail_vals:
-        while (n_contacts-- > 0)
+        while (contacts != NULL && n_contacts-- > 0)
                 dht_contact_msg__free_unpacked(contacts[n_contacts], NULL);
+
         free(contacts);
  fail_contacts:
         return NULL;
@@ -3312,6 +3316,42 @@ static int emergency_peer(struct list_head * pl)
         return -ENOMEM;
 }
 
+static bool __dht_kv_bucket_has_addr(struct bucket * b,
+                                     uint64_t        addr)
+{
+        struct list_head * p;
+        size_t             i;
+
+        assert(b != NULL);
+
+        if (*b->children != NULL)
+                for (i = 0; i < (1L << DHT_BETA); ++i)
+                        if (__dht_kv_bucket_has_addr(b->children[i], addr))
+                                return true;
+
+        llist_for_each(p, &b->contacts) {
+                struct contact * c;
+                c = list_entry(p, struct contact, next);
+                if (c->addr == addr)
+                        return true;
+        }
+
+        return false;
+}
+
+static bool dht_kv_knows_peer(void)
+{
+        bool found;
+
+        pthread_rwlock_rdlock(&dht.db.lock);
+
+        found = __dht_kv_bucket_has_addr(dht.db.contacts.root, dht.peer);
+
+        pthread_rwlock_unlock(&dht.db.lock);
+
+        return found;
+}
+
 static int dht_kv_seed_bootstrap_peer(void)
 {
         struct list_head pl;
@@ -3323,6 +3363,9 @@ static int dht_kv_seed_bootstrap_peer(void)
                 return 0;
         }
 
+        if (dht_kv_knows_peer())
+                return 0;
+
         if (emergency_peer(&pl) < 0) {
                 log_err("Could not create emergency peer.");
                 goto fail_peer;
@@ -3338,7 +3381,8 @@ static int dht_kv_seed_bootstrap_peer(void)
 
         peer_list_destroy(&pl);
 
-        return 0;
+        /* Sent, but not bootstrapped until the peer is in the DHT. */
+        return -EAGAIN;
  fail_query:
         peer_list_destroy(&pl);
  fail_peer:
@@ -3427,6 +3471,8 @@ static void value_list_destroy(struct list_head * vl)
 #define MUST_REPLICATE(v, now) ((now)->tv_sec > (v)->t_repl + dht.t_repl)
 #define MUST_REPUBLISH(v, now) /* Close to expiry deadline */ \
         (((v)->t_exp - (now)->tv_sec) < (DHT_N_REPUB * dht.t_repl))
+/* A local value must be (re)stored if near expiry or never stored. */
+#define MUST_STORE_LVAL(v, now) (MUST_REPUBLISH(v, now) || (v)->t_repl == 0)
 static void dht_entry_get_repl_lists(const struct dht_entry * e,
                                      struct list_head *       repl,
                                      struct list_head *       rebl,
@@ -3448,7 +3494,7 @@ static void dht_entry_get_repl_lists(const struct dht_entry * e,
 
         llist_for_each(p, &e->lvals) {
                 struct val_entry * v = list_entry(p, struct val_entry, next);
-                if (MUST_REPLICATE(v, now) && MUST_REPUBLISH(v, now)) {
+                if (MUST_REPLICATE(v, now) && MUST_STORE_LVAL(v, now)) {
                         /* Add expire time here, to allow creating val_entry */
                         n = val_entry_create(v->val, now->tv_sec + dht.t_exp);
                         if (n == NULL)
@@ -3738,6 +3784,9 @@ static void * work(void * o)
 
         log_dbg("DHT worker starting %ld seconds interval.", intv * n);
 
+        /* Flush names registered before we had peers to store them. */
+        dht_kv_replicate();
+
         while (true) {
                 int i = 0;
                 while (tasks[i] != NULL) {