DAOS-13938 dfuse: adjust offset in readdir cache entry list (#15190)

The dfuse readdir caching feature stats entries returned by
dfs readdir that might be legimitely deleted by another process
or another node. When this happens, the dfuse code decides to
skip the deleted entry, but might leave the readdir entry cache
in an inconsistent state causing the following assertion failure
in dfuse_dre_drop():
drc->drc_next_offset == expected_offset + 1 || drc->drc_next_offset == READDIR_EOD

This patch fixes the offset in the linked list of cached readdir
entries to support skipped entries.

Signed-off-by: Johann Lombardi <johann.lombardi@gmail.com>

src/client/dfuse/ops/readdir.c

@@ -133,7 +133,7 @@ dfuse_dre_drop(struct dfuse_info *dfuse_info, struct dfuse_obj_hdl *oh)
 	struct dfuse_readdir_hdl *hdl;
 	struct dfuse_readdir_c   *drc, *next;
 	uint32_t                  oldref;
-	off_t                     expected_offset = 2;
+	off_t                     next_offset = 0;
 
 	DFUSE_TRA_DEBUG(oh, "Dropping ref on %p", oh->doh_rd);
 
@@ -161,10 +161,11 @@ dfuse_dre_drop(struct dfuse_info *dfuse_info, struct dfuse_obj_hdl *oh)
 		oh->doh_ie->ie_rd_hdl = NULL;
 
 	d_list_for_each_entry_safe(drc, next, &hdl->drh_cache_list, drc_list) {
-		D_ASSERT(drc->drc_offset == expected_offset);
-		D_ASSERT(drc->drc_next_offset == expected_offset + 1 ||
-			 drc->drc_next_offset == READDIR_EOD);
-		expected_offset = drc->drc_next_offset;
+		/** Verify offset/next_offset are consistent in the entry cache list */
+		D_ASSERTF(next_offset == 0 || next_offset == drc->drc_offset,
+			  "Inconsistent offset list %#lx prev next %#lx " DF_DE,
+			  drc->drc_offset, next_offset, DP_DE(drc->drc_name));
+		next_offset = drc->drc_next_offset;
 		if (drc->drc_rlink)
 			d_hash_rec_decref(&dfuse_info->dpi_iet, drc->drc_rlink);
 		D_FREE(drc);
@@ -472,7 +473,7 @@ dfuse_do_readdir(struct dfuse_info *dfuse_info, fuse_req_t req, struct dfuse_obj
 
 					if (rc != 0) {
 						DFUSE_TRA_DEBUG(oh, "Problem finding file %d", rc);
-						D_GOTO(reply, 0);
+						D_GOTO(reply, rc);
 					}
 					/* Check oid is the same! */
 
@@ -689,7 +690,8 @@ dfuse_do_readdir(struct dfuse_info *dfuse_info, fuse_req_t req, struct dfuse_obj
 			if (rc == ENOENT) {
 				DFUSE_TRA_DEBUG(oh, "File does not exist");
 				D_FREE(drc);
-				continue;
+				/** legitimate race */
+				D_GOTO(next, rc = 0);
 			} else if (rc != 0) {
 				DFUSE_TRA_DEBUG(oh, "Problem finding file %d", rc);
 				D_FREE(drc);
@@ -759,6 +761,12 @@ dfuse_do_readdir(struct dfuse_info *dfuse_info, fuse_req_t req, struct dfuse_obj
 			}
 
 			if (drc) {
+				if (oh->doh_rd_nextc != NULL)
+					/**
+					 * Fix next offset of previous cache entry in case some
+					 * dre entries were skipped (e.g. failed stat call).
+					 */
+					oh->doh_rd_nextc->drc_next_offset = drc->drc_offset;
 				oh->doh_rd_nextc = drc;
 				DFUSE_TRA_DEBUG(hdl, "Appending offset %#lx to list, next %#lx",
 						drc->drc_offset, drc->drc_next_offset);
@@ -769,12 +777,15 @@ dfuse_do_readdir(struct dfuse_info *dfuse_info, fuse_req_t req, struct dfuse_obj
 			dre->dre_offset = 0;
 			buff_offset += written;
 			added++;
+next:
 			offset++;
 			oh->doh_rd_offset = dre->dre_next_offset;
 
 			if (dre->dre_next_offset == READDIR_EOD) {
 				DFUSE_TRA_DEBUG(oh, "Reached end of directory");
 				oh->doh_rd_offset = READDIR_EOD;
+				if (hdl->drh_caching && oh->doh_rd_nextc != NULL)
+					oh->doh_rd_nextc->drc_next_offset = READDIR_EOD;
 				D_GOTO(reply, rc = 0);
 			}
 		}