Fix issue with ESP-NOW, improve logging and subscription recovery

Author: nicklasb

components/robusto/base/src/logging/logging_robusto.c

@@ -48,6 +48,9 @@ void rob_log_write(rob_log_level_t level,
     va_end(list);
 }
 #endif
+
+// TODO: Add a Mac address log function
+
 // TODO: Move rob_log_bit_mesh into a macro instead so we don't need this.
 void rob_log_bit_mesh(rob_log_level_t level,
                       const char *tag,

components/robusto/include/robusto_pubsub_client.h

@@ -21,6 +21,7 @@ typedef enum
     TOPIC_STATE_INACTIVE, /* The topic in inactive */
     TOPIC_STATE_ACTIVE, /* The topic is inactive */
     TOPIC_STATE_PROBLEM, /* The topic has some problems */
+    TOPIC_STATE_RECOVERING, /* We have recently published data */
     TOPIC_STATE_UNKNOWN, /* The topic is unknown by the server */
     TOPIC_STATE_PUBLISHED, /* We have recently published data */
     TOPIC_STATE_STALE, /* Haven't gotten or published data in stale_time_ms */
@@ -46,8 +47,6 @@ struct subscribed_topic
     uint32_t stale_time_ms;
     /* Calculated hash used for referencing the topic */
     uint32_t topic_hash;
-    /* Is being restored*/
-    bool restoring;
     /* The peer */
     robusto_peer_t *peer;
     /* A conversation ID for communication */

components/robusto/misc/src/pubsub/robusto_pubsub_client.c

@@ -154,7 +154,7 @@ void incoming_callback(robusto_message_t *message)
             ROB_LOGW(pubsub_client_log_prefix, "Server told us that the topic %s (topic hash %lu) is unknown,", curr_topic->topic_name, curr_topic->topic_hash);
             set_topic_state(curr_topic, TOPIC_STATE_UNKNOWN);
         } else {
-            ROB_LOGW(pubsub_client_log_prefix, "Server told us that the %lu topic hash is unknown, it is to us too, newly removed?", *(uint32_t *)(message->binary_data + 1));
+            ROB_LOGW(pubsub_client_log_prefix, "Server told us that the %lu topic hash is unknown, it is to us too, newly removed?", curr_topic->topic_hash);
         }
 
     } else
@@ -255,7 +255,6 @@ subscribed_topic_t *_add_topic_and_conv(robusto_peer_t *peer, char *topic_name,
     new_topic->next = NULL;
     new_topic->peer = peer;
     new_topic->topic_hash = 0;
-    new_topic->restoring = false;
     new_topic->callback = callback;
     new_topic->display_offset = display_offset;
     new_topic->state = TOPIC_STATE_UNSET;
@@ -302,7 +301,7 @@ subscribed_topic_t *robusto_pubsub_client_get_topic(robusto_peer_t *peer, char *
         msg[0] = PUBSUB_GET_TOPIC;
     }
 
-    ROB_LOGE(pubsub_client_log_prefix, "Sending subscription for %s conversation_id %u!", topic_name, new_topic->conversation_id);
+    ROB_LOGE(pubsub_client_log_prefix, "Sending subscription for %s conversation_id %u, hash: %lu", topic_name, new_topic->conversation_id, new_topic->topic_hash);
     rob_ret_val_t ret_sub = send_message_binary(peer, PUBSUB_SERVER_ID, new_topic->conversation_id, (uint8_t *)msg, data_length, NULL);
     if (ret_sub != ROB_OK) {
         ROB_LOGE(pubsub_client_log_prefix, "Pub Sub client: Subscription failed, failed to queue or send message %s.", new_topic->topic_name);
@@ -327,18 +326,28 @@ void recover_topic(subscribed_topic_t *topic)
     if (topic->state == TOPIC_STATE_PROBLEM && !robusto_waitfor_byte_change(&topic->state, 1000) != ROB_OK)
     {
         ROB_LOGE(pubsub_client_log_prefix, "Failed to recover %s using the %s peer", topic->topic_name, topic->peer->name);
+        topic->state = TOPIC_STATE_PROBLEM;
+        r_delay(5000);
     }
     else
     {
         // A successful incoming call will set the state to inactive or active elsewhere
         ROB_LOGI(pubsub_client_log_prefix, "Recovered %s using the %s peer!", topic->topic_name, topic->peer->name);
+        topic->state = TOPIC_STATE_ACTIVE;
     }
-    topic->restoring = false;
+    
     robusto_delete_current_task();
 }
 
 void create_topic_recovery_task(subscribed_topic_t *topic)
 {
+    topic->state = TOPIC_STATE_RECOVERING;
+
+    if (topic->peer->state < PEER_KNOWN_INSECURE){
+       ROB_LOGW(pubsub_client_log_prefix, "Not creating a topic recovery task for %s topic, peer %s because the peer is not working properly.",
+             topic->topic_name, topic->peer->name); 
+        return;
+    }
     ROB_LOGW(pubsub_client_log_prefix, "Creating a topic recovery task for %s topic, peer %s",
              topic->topic_name, topic->peer->name);
     char *task_name;
@@ -355,7 +364,7 @@ void robusto_pubsub_check_topics()
     subscribed_topic_t *curr_topic = first_subscribed_topic;
     while (curr_topic)
     {
-        if (curr_topic->restoring)
+        if (curr_topic->state == TOPIC_STATE_RECOVERING)
         {
             // Do nothing regardless of state to not disturb any existing recovery processes
         }
@@ -365,7 +374,7 @@ void robusto_pubsub_check_topics()
             if (curr_topic->peer->problematic_media_types != curr_topic->peer->supported_media_types)
             {
                 create_topic_recovery_task(curr_topic);
-                r_delay(1000);
+                r_delay(5000);
             }
         }
 

components/robusto/misc/src/pubsub/robusto_pubsub_server.c

@@ -97,7 +97,7 @@ uint32_t robusto_pubsub_add_subscriber(pubsub_server_topic_t *topic, robusto_pee
     pubsub_server_subscriber_t * existing = find_subscription_by_subscriber_hash(topic, peer);
     // TODO: Handle cases where adding with only callback, they might be double
     if (existing) {
-        ROB_LOGW(pubsub_log_prefix, "%s asked to subscribe to %s, but is already a subscriber, returning hash.", peer->name, topic->name);
+        ROB_LOGW(pubsub_log_prefix, "%s asked to subscribe to %s, but is already a subscriber, returning hash: %lu.", peer->name, topic->name, topic->hash);
         return topic->hash;
     }
 

components/robusto/network/src/incoming/incoming.c

@@ -48,6 +48,7 @@ static incoming_callback_cb *incoming_callback = NULL;
 rob_ret_val_t robusto_handle_incoming(uint8_t *data, uint32_t data_length, robusto_peer_t *peer, e_media_type media_type, int offset)
 {
     // TODO: Messages might be too large for SRAM and might need to be stored elsewhere (flash), how to handle? We might also have to *send* large files.
+    // TODO: As this might be done inside receive callback, it should possibly spawn a task instead
     // rob_log_bit_mesh(ROB_LOG_WARN, incoming_log_prefix, data + offset, data_length - offset);
     ROB_LOGD(incoming_log_prefix, "In robusto_handle_incoming.");
     robusto_message_t *message;

components/robusto/network/src/media/espnow/espnow_control.c

@@ -50,7 +50,8 @@ static char *espnow_log_prefix;
 
 void init_wifi()
 {
-
+    ROB_LOGI(espnow_log_prefix, "Creating default event loop.");
+    ESP_ERROR_CHECK(esp_event_loop_create_default());
     ROB_LOGI(espnow_log_prefix, "Initializing wifi (for ESP-NOW)");
     ESP_ERROR_CHECK(esp_netif_init());
     ROB_LOGI(espnow_log_prefix, "esp_netif_init done.");

components/robusto/network/src/media/espnow/espnow_messaging.c

@@ -52,13 +52,12 @@
 
 static char *espnow_log_prefix;
 
-#define ESPNOW_MAXDELAY 512
 #define ESPNOW_FRAGMENT_SIZE (ESP_NOW_MAX_DATA_LEN - 10)
 
 static void espnow_deinit(espnow_send_param_t *send_param);
 
 static bool has_receipt = false;
-static esp_now_send_status_t send_status = ESP_NOW_SEND_FAIL;
+static int send_status = -1;
 
 // For synchronously reading async data
 static uint8_t *synchro_data = NULL;
@@ -69,9 +68,12 @@ rob_ret_val_t esp_now_send_check(robusto_peer_t *peer, uint8_t *data, uint32_t d
 {
 
     // TODO: Recommend more wifi TX buffers and add warning if not enabled
-    has_receipt = false;
+    
     ROB_LOGD(espnow_log_prefix, "esp_now_send_check, sending %lu bytes.", data_length);
     int rc = esp_now_send(&peer->base_mac_address, data, data_length);
+    send_status = -1;
+    has_receipt = false;
+
     if (rc != ESP_OK)
     {
         ROB_LOGE(espnow_log_prefix, "Mac address:");
@@ -90,7 +92,8 @@ rob_ret_val_t esp_now_send_check(robusto_peer_t *peer, uint8_t *data, uint32_t d
         }
         else if (rc == ESP_ERR_ESPNOW_NO_MEM)
         {
-            ROB_LOGE(espnow_log_prefix, "ESP-NOW error: ESP_ERR_ESPNOW_NO_MEM");
+            ROB_LOGE(espnow_log_prefix, "ESP-NOW error: ESP_ERR_ESPNOW_NO_MEM - Will delay a short while to let it free its memory.");
+            r_delay(3000);
         }
         else if (rc == ESP_ERR_ESPNOW_FULL)
         {
@@ -121,15 +124,29 @@ rob_ret_val_t esp_now_send_check(robusto_peer_t *peer, uint8_t *data, uint32_t d
         rc = ROB_OK;
         add_to_history(&peer->espnow_info, true, rc);
     }
+    // We will wait here until the message was sent, for 2 seconds
+    int32_t start_send = r_millis();
+    while ((send_status < 0) && (r_millis() < start_send + 2000))
+    {
+        robusto_yield();
+    }
 
+    if (send_status == ESP_NOW_SEND_FAIL) {
+        ROB_LOGE(espnow_log_prefix, "ESP-NOW transmission failed completing after %lu ms. Peer: %s", r_millis() - start_send, peer->name);
+        return ROB_FAIL;
+    }
+    if (send_status < 0) {
+        ROB_LOGE(espnow_log_prefix, "ESP-NOW transmission didn't complete in time (2000 ms). Peer: %s", peer->name);
+        return ROB_FAIL;
+    }
     if (!receipt)
     {
         return rc;
     }
 
-    // We want to wait to make sure the transmission is done.
+    // We want to wait to make sure the transmission is received successfully.
     int32_t start = r_millis();
-    while ((!has_receipt) && (r_millis() < start + 2000))
+    while ((!has_receipt) && (r_millis() < start + 500))
     {
         robusto_yield();
     }
@@ -281,13 +298,14 @@ static void espnow_recv_cb(const esp_now_recv_info_t *esp_now_info, const uint8_
             has_receipt = true;
             return;
         }
-
         // Send a receipt
         uint8_t response[2];
         response[0] = 0xff;
         response[1] = 0x00;
-
-        esp_now_send_check(peer, &response, 2, false);
+        // NOTE: We are calling esp_now_send directly here as we are calling it inside of the receive callback
+        if (esp_now_send(&peer->base_mac_address, &response, 2) != ROB_OK) {
+            ROB_LOGE(espnow_log_prefix, ">> espnow_recv_cb failed to send a receipt to %s.", peer->name);
+        }
         if (!handled)
         {
             // Copy data a ESP-NOW frees it

components/robusto/network/src/message/robusto_message_sending.c

@@ -308,16 +308,19 @@ void send_work_item(media_queue_item_t * queue_item, robusto_media_t *info, e_me
 
         add_to_history(info, true, retval);
     } else {
-        ROB_LOGI(message_sending_log_prefix, ">> As the %s, mt %i is recovering, we need to try some other media for a message (%i, %i)", queue_item->peer->name, media_type, info->state, queue_item->queue_item_type );
+        
+        ROB_LOGI(message_sending_log_prefix, ">> As the %s, mt %i is recovering, we might need to try some other media for a message (%i, %i)", queue_item->peer->name, media_type, info->state, queue_item->queue_item_type );
     }
 
     if ((retval != ROB_OK) && // We only try other medias if we have failed..
     (queue_item->receipt) && // ..and if it is receipt required, then we infer that we will try with multiple medias
-    (retval != ROB_ERR_WHO)) // ..or if the peer knew who we were (it will only respond to unknowns on wired connections)
+    (retval != ROB_ERR_WHO) && // ..or if the peer knew who we were (it will only respond to unknowns on wired connections)
+    (get_host_supported_media_types() != (queue_item->exclude_media | media_type))) // And that there are no other medias to try
     {
         e_media_type next_media_type;
         // Add current media type to excluded media
         queue_item->exclude_media = queue_item->exclude_media | media_type;
+
         ROB_LOGW(message_sending_log_prefix, "Failed sending using %s, will try some other media.", media_type_to_str(media_type));
         // Check suitability again to find another media to try
         rob_ret_val_t suitability_res = set_suitable_media(queue_item->peer, queue_item->data_length, queue_item->exclude_media, &next_media_type);

components/robusto/network/src/qos/qos_recovery.c

@@ -78,7 +78,9 @@ void task_recover(recover_params_t *params)
                 set_state(peer, params->info, media_type, media_state_working, media_problem_none);
                 rec_retval = ROB_OK;
             } else {
+                set_state(peer, params->info, media_type, media_state_problem, media_problem_unknown);
                 ROB_LOGW(recovery_log_prefix, ">> Sending presentation failed");
+                rec_retval = ROB_FAIL;
             }
         }
         params->info->postpone_qos = false;
@@ -88,7 +90,7 @@ void task_recover(recover_params_t *params)
             send_heartbeat_message(peer, media_type);
         }
     }
-
+    
     robusto_delete_current_task();
 }
 

components/robusto/network/src/qos/qos_state.c

@@ -133,9 +133,14 @@ void working_state(robusto_peer_t *peer, robusto_media_t *info, uint64_t last_he
 // Problem state behavior
 void problem_state(robusto_peer_t *peer, robusto_media_t *info, uint64_t last_heartbeat_time, e_media_type media_type)
 {
-    // Has it been having a problem more than one cycle, mark it as as recovering.
+    
+    if (info->state == media_state_recovering) {
+        // If is in recovery, do nothing.
+        return;
+    }
     if (info->last_state_change < last_heartbeat_time - 1000)
     {
+        // Has it been having a problem more than one cycle, mark it as as recovering and initiate recovery
         ROB_LOGE(qos_state_log_prefix, "%s, media type %s has had a problem for too long, going into recovery.", peer->name, media_type_to_str(media_type));
         set_state(peer, info, media_type, media_state_recovering, info->problem);
         recover_media(peer, info, last_heartbeat_time, media_type);