fellow_cache: Implement object deletion for FCO_READFAIL

src/fellow_cache.c

@@ -5944,6 +5944,21 @@ fellow_cache_obj_wait_written(struct fellow_cache_obj *fco)
 	AZ(pthread_mutex_unlock(&fco->mtx));
 }
 
+static void
+fellow_disk_obj_delete_thin(struct fellow_fd *ffd,
+    fellow_disk_block start, const uint8_t hash[DIGEST_LEN])
+{
+	static const char ndle = 1;
+	struct fellow_dle e[ndle];
+
+	fellow_dle_init(e, ndle);
+	e->type = DLEDSK(DLE_OBJ_DEL_THIN);
+	memcpy(e->u.obj.hash, hash, (size_t)DIGEST_LEN);
+	e->u.obj.start = start;
+
+	fellow_log_dle_submit(ffd, e, ndle);
+}
+
 /* evict, then delete from log (this order is important to avoid
  * fdb reuse)
  *
@@ -5962,6 +5977,7 @@ fellow_cache_obj_delete(struct fellow_cache *fc,
 	struct buddy_returns *rets;
 	struct buddy *dskbuddy;
 	struct stvfe_tune *tune;
+	enum fcos_state state;
 	enum fcol_state logstate;
 	fellow_disk_block fdba;
 	unsigned u, n;
@@ -6017,6 +6033,7 @@ fellow_cache_obj_delete(struct fellow_cache *fc,
 	}
 
 	logstate = fco->logstate;
+	state = FCO_STATE(fco);
 
 	// must have been the last reference
 	AZ(fellow_cache_obj_deref_locked(lcb, fc, fco));
@@ -6029,19 +6046,34 @@ fellow_cache_obj_delete(struct fellow_cache *fc,
 	case FCOL_TOOLATE:
 		break;
 	case FCOL_DELETED:
-		fellow_disk_obj_delete_submit(fc->ffd, fdba, hash,
-		    region, n);
+		/* if read failed, we do not have all regions, but the log might
+		 * still work */
+
+		if (state == FCO_READFAIL)
+			fellow_disk_obj_delete_thin(fc->ffd, fdba, hash);
+		else {
+			fellow_disk_obj_delete_submit(fc->ffd, fdba, hash,
+			    region, n);
+		}
 		return;
 	default:
 		WRONG("fellow_cache_obj_delete logstate UNLOCKED");
 	}
 
+	/*
+	 * NOTE: FOR FCO_READFAIL, we are likely to leak disk space
+	 *
+	 * When fellow_obj_regions() fails reading a segment list, we miss
+	 * segments and, consequently, regions which the object occupies.
+	 */
+
 	/* XXX no async IO here yet: We need to free the dskbuddy space after
 	 * the discard is complete. To do that, we would need to store the
 	 * off/size of the discard in an additional structure of the io ring.
 	 *
 	 * we can not just hold the cache async io mtx because it would delay
 	 * readahead
+	 *
 	 */
 
 	assert(n < FCO_MAX_REGIONS + 1);