app: storage + cloud: Keep data in storage until cloud confirms sent

Modify storage module to keep data in the backend until cloud confirms
that an item has been sent. This allows for better handling of failed
sends, as the data will still be available in storage for retrying.

Signed-off-by: Trond F. Christiansen <trond.christiansen@nordicsemi.no>

Author: Trond-F-Christiansen

app/src/modules/cloud/cloud.c

@@ -525,6 +525,27 @@ static int request_storage_batch_data(uint32_t session_id)
 	return 0;
 }
 
+static bool handle_send_error(int err, struct storage_data_item *item)
+{
+	/* Non-network errors (ENOTSUP, EINVAL, ENODATA) indicate malformed or
+	 * unsupported data that will never succeed.
+	 */
+	if (err == -ENOTSUP || err == -EINVAL || err == -ENODATA) {
+		LOG_ERR("Data error sending storage data (type %d), "
+			"dropping item: %d",
+			item->type, err);
+
+		return false;
+	}
+
+	/* Network/transport error */
+	LOG_ERR("Network error sending storage data (type %d), "
+		"aborting session: %d",
+		item->type, err);
+
+	return true;
+}
+
 static void handle_storage_batch_available(const struct storage_msg *msg)
 {
 	int err;
@@ -536,10 +557,22 @@ static void handle_storage_batch_available(const struct storage_msg *msg)
 		.type = STORAGE_BATCH_CLOSE,
 		.session_id = session_id,
 	};
+	struct storage_msg consume_msg = {
+		.type = STORAGE_BATCH_CONSUME,
+		.session_id = session_id,
+	};
 	bool session_error = false;
 
 	LOG_INF("Processing storage batch: %u items available", items_available);
 
+	/* Suppress delivery of storage_chan messages back to cloud_subscriber while we
+	 * are blocking in this loop.
+	 */
+	err = zbus_obs_set_chan_notification_mask(&cloud_subscriber, &storage_chan, true);
+	if (err) {
+		LOG_WRN("Failed to mask storage_chan for cloud_subscriber, error: %d", err);
+	}
+
 	/* Drain the batch buffer: read until timeout, abort on hard error */
 	while (!session_error) {
 		err = storage_batch_read(&item, K_MSEC(500));
@@ -557,14 +590,34 @@ static void handle_storage_batch_available(const struct storage_msg *msg)
 
 		err = send_storage_data_to_cloud(&item);
 		if (err) {
-			LOG_ERR("Failed to send storage data to cloud, error: %d", err);
+			session_error = handle_send_error(err, &item);
+			if (session_error) {
+				/* Network error, abort */
+				continue;
+			}
+			/* For data errors, fall through to consume the item and prevent retrying
+			 * it in the next session.
+			 */
+		} else {
+			items_processed++;
 		}
 
-		items_processed++;
+		/* Consume the item: confirms a successful send or skips a malformed item */
+		consume_msg.data_type = item.type;
+		err = zbus_chan_pub(&storage_chan, &consume_msg, PUB_TIMEOUT);
+		if (err) {
+			LOG_ERR("Failed to consume storage item, error: %d", err);
+		}
 	}
 
 	LOG_DBG("Processed %u/%u storage items", items_processed, items_available);
 
+	/* Re-enable storage_chan notifications to cloud_subscriber */
+	err = zbus_obs_set_chan_notification_mask(&cloud_subscriber, &storage_chan, false);
+	if (err) {
+		LOG_WRN("Failed to unmask storage_chan for cloud_subscriber, error: %d", err);
+	}
+
 	if (!session_error && msg->more_data) {
 		LOG_DBG("More data available in batch, requesting next batch");
 

app/src/modules/storage/Kconfig.storage

@@ -88,7 +88,7 @@ config APP_STORAGE_BATCH_BUFFER_SIZE
 
 config APP_STORAGE_SESSION_TIMEOUT_SECONDS
 	int "Storage batch session timeout"
-	default 60
+	default 30
 	help
 	  Timeout for an active storage batch session. If no activity
 	  occurs on the session for this duration, it will be automatically

app/src/modules/storage/storage.c

@@ -480,110 +480,103 @@ static void send_batch_available_response(uint32_t session_id, size_t item_count
 	send_batch_response(STORAGE_BATCH_AVAILABLE, session_id, item_count, more_available);
 }
 
-/* Populate the pipe with all stored data
+/* Populate the pipe with the next pending item.
  *
- * @return 0 on success (all data written or pipe full)
- * @return -EIO on data retrieval or size validation error
+ * Peeks the head item of the first non-empty type and writes it to the pipe.
+ * The item is NOT removed from the backend; that only happens on STORAGE_BATCH_CONSUME.
+ *
+ * @return 0 on success (one item written to pipe)
+ * @return -ENODATA if no items are available across all types
+ * @return -EIO on peek or pipe write error
  */
 static int populate_pipe(struct storage_state *state_object)
 {
 	const struct storage_backend *backend = storage_backend_get();
-	size_t total_bytes_sent = 0;
 
 	state_object->current_session.more_data = false;
 
-	/* Populate pipe with all stored data */
 	STRUCT_SECTION_FOREACH(storage_data, type) {
-		int count = backend->count(type);
-
-		while (count > 0) {
-			int ret;
-			size_t total_size;
-			uint8_t item_buffer[sizeof(struct storage_pipe_header) +
-					    STORAGE_MAX_DATA_SIZE];
-			struct storage_pipe_header *header =
-				(struct storage_pipe_header *)item_buffer;
-			uint8_t *data = item_buffer + sizeof(struct storage_pipe_header);
-
-			/* Peek at size without copying data (data = NULL) */
-			ret = backend->peek(type, NULL, 0);
-			if (ret == -EAGAIN) {
-				/* No more data of this type */
-				break;
-			} else if (ret < 0) {
-				LOG_ERR("Failed to peek %s data size: %d", type->name, ret);
-
-				return -EIO;
-			}
-
-			/* Prepare header with actual size */
-			header->type = (uint8_t)type->data_type;
-
-			if ((ret < 0) || (ret > (int)STORAGE_MAX_DATA_SIZE) ||
-			    (ret > UINT16_MAX)) {
-				LOG_ERR("Invalid data size for header: %d", ret);
-
-				return -EIO;
-			}
+		int ret;
+		uint8_t item_buffer[sizeof(struct storage_pipe_header) + STORAGE_MAX_DATA_SIZE];
+		struct storage_pipe_header *header = (struct storage_pipe_header *)item_buffer;
+		uint8_t *data = item_buffer + sizeof(struct storage_pipe_header);
+		size_t total_size;
+
+		/* Peek the head item */
+		ret = backend->peek(type, data, STORAGE_MAX_DATA_SIZE);
+		if (ret == -EAGAIN) {
+			/* No items of this type, try next */
+			continue;
+		} else if (ret < 0) {
+			LOG_ERR("Failed to peek %s data: %d", type->name, ret);
+			return -EIO;
+		}
 
-			header->data_size = (uint16_t)ret;
+		if (ret > UINT16_MAX) {
+			LOG_ERR("Peek returned oversized item for %s: %d", type->name, ret);
+			return -EIO;
+		}
 
-			/* Calculate exact total size needed using actual data size */
-			total_size = sizeof(struct storage_pipe_header) + ret;
-			if (total_size > sizeof(item_buffer)) {
-				LOG_ERR("Combined data too large: %zu > %zu",
-					total_size, sizeof(item_buffer));
+		header->type = (uint8_t)type->data_type;
+		header->data_size = (uint16_t)ret;
+		total_size = sizeof(struct storage_pipe_header) + (size_t)ret;
 
-				return -EIO;
-			}
+		ret = pipe_write_all(&storage_pipe, item_buffer, total_size, PUB_TIMEOUT);
+		if (ret < 0) {
+			LOG_ERR("Unexpected pipe write failure: %d", ret);
+			return -EIO;
+		}
 
-			/* Check if exact size fits in remaining pipe buffer space */
-			if (total_bytes_sent + total_size > CONFIG_APP_STORAGE_BATCH_BUFFER_SIZE) {
-				/* Pipe buffer full - stop here without consuming data */
-				LOG_DBG("Pipe buffer full");
+		state_object->current_session.items_sent++;
 
-				state_object->current_session.more_data = true;
+		LOG_DBG("Pipe populated for session 0x%X: with %s item (%zu bytes)",
+			state_object->current_session.session_id,
+			type->name, total_size);
 
-				break;
-			}
+		return 0;
+	}
 
-			/* Now that we know it fits, retrieve the data from backend */
-			ret = backend->retrieve(type, data, STORAGE_MAX_DATA_SIZE);
-			if (ret < 0) {
-				LOG_ERR("Failed to retrieve %s data after peek: %d",
-					type->name, ret);
+	return -ENODATA;
+}
 
-				return -EIO;
-			}
+/* Consume the confirmed-sent item from the backend and peek the next one into the pipe.
+ * Returns false if the session_id does not match the active session (message ignored).
+ */
+static bool handle_batch_consume(struct storage_state *state_object,
+				 const struct storage_msg *msg)
+{
+	const struct storage_backend *backend = storage_backend_get();
+	uint8_t discard[STORAGE_MAX_DATA_SIZE];
+	int ret;
 
-			/* Sanity check: retrieved size should match peeked size */
-			__ASSERT_NO_MSG(ret == (int)header->data_size);
+	if (msg->session_id != state_object->current_session.session_id) {
+		LOG_WRN("CONSUME session ID mismatch: 0x%X (current: 0x%X), ignoring",
+			msg->session_id, state_object->current_session.session_id);
+		return false;
+	}
 
-			/* Write combined buffer atomically to pipe */
-			ret = pipe_write_all(&storage_pipe, item_buffer, total_size, PUB_TIMEOUT);
+	/* Remove the confirmed-sent item from the backend queue head */
+	STRUCT_SECTION_FOREACH(storage_data, type) {
+		if (type->data_type == msg->data_type) {
+			ret = backend->retrieve(type, discard, sizeof(discard));
 			if (ret < 0) {
-				/* This should never happen since we checked space above */
-				LOG_ERR("Unexpected pipe write failure after space check: %d", ret);
-
-				return -EIO;
+				LOG_ERR("Failed to consume item (type %d): %d", msg->data_type, ret);
+			} else {
+				LOG_DBG("Consumed %s item from backend", type->name);
 			}
-
-			__ASSERT_NO_MSG(ret == (int)total_size);
-
-			/* Update session progress and byte tracking */
-			state_object->current_session.items_sent++;
-			total_bytes_sent += total_size;
-
-			count--;
+			break;
 		}
 	}
 
-	LOG_DBG("Batch population complete for session 0x%X: %zu/%zu items",
-		state_object->current_session.session_id,
-		state_object->current_session.items_sent,
-		state_object->current_session.total_items);
+	/* Peek the next item into the pipe for the consumer */
+	ret = populate_pipe(state_object);
+	if (ret == -ENODATA) {
+		LOG_DBG("All items consumed, pipe empty");
+	} else if (ret < 0) {
+		LOG_ERR("Failed to populate pipe after consume: %d", ret);
+	}
 
-	return 0;
+	return true;
 }
 
 /* Start a new batch session.
@@ -633,10 +626,10 @@ static int start_batch_session(struct storage_state *state_object,
 		return err;
 	}
 
-	/* Success - pipe populated (fully or partially) */
+	/* Success, one item in pipe, report total available to consumer */
 	send_batch_available_response(request_msg->session_id,
-				      state_object->current_session.items_sent,
-				      state_object->current_session.more_data);
+				      total_items,
+				      false);
 
 	LOG_DBG("Started batch session (session_id 0x%X), %zu items in batch (%zu total)",
 		state_object->current_session.session_id,
@@ -881,6 +874,13 @@ static enum smf_state_result state_buffer_pipe_active_run(void *o)
 
 			return SMF_EVENT_HANDLED;
 
+		case STORAGE_BATCH_CONSUME:
+			if (handle_batch_consume(state_object, msg)) {
+				k_work_reschedule(&state_object->session_timeout_work,
+						  K_SECONDS(STORAGE_SESSION_TIMEOUT_SECONDS));
+			}
+			return SMF_EVENT_HANDLED;
+
 		case STORAGE_BATCH_REQUEST: {
 			int err;
 

app/src/modules/storage/storage.h

@@ -82,6 +82,13 @@ enum storage_msg_type {
 
 	/* Batch is busy - cannot process request at this time. */
 	STORAGE_BATCH_BUSY,
+
+	/* Confirm that a batch item was successfully sent to cloud.
+	 * Must contain `data_type` identifying the type of the item just sent.
+	 * Storage removes the item from the backend queue head and makes the next
+	 * item available in the pipe. Only valid during an active batch session.
+	 */
+	STORAGE_BATCH_CONSUME,
 };
 
 /**