fix(lsm): remove data file for new entries that are removed before being

stored on disk

lsm/src/_include/lsm/store_internal.h

@@ -156,4 +156,11 @@ lsm_error lsm_entry_data_open_read(lsm_entry_handle *handle);
  */
 lsm_error lsm_entry_data_open_write(lsm_entry_handle *handle);
 
+/**
+ * Remove the entry's data file if present and close its handle.
+ *
+ * @param handle handle to the entry
+ */
+lsm_error lsm_entry_data_remove(lsm_entry_handle *handle);
+
 #endif

lsm/src/store/lsm_store_disk_write.c

@@ -153,20 +153,7 @@ lsm_error lsm_entry_disk_remove(lsm_entry_handle *handle) {
 
   fflush(store->idx.f);
 
-  // Remove data file if present
+  LSM_RES(lsm_entry_data_remove(handle));
-  if (entry->data_len > 0) {
-    if (handle->f != NULL) {
-      fclose(handle->f);
-      handle->f = NULL;
-    }
-
-    char data_path[lsm_entry_data_path_len(handle) + 1];
-    lsm_entry_data_path(data_path, handle);
-
-    if (remove(data_path) != 0) {
-      return lsm_error_failed_io;
-    }
-  }
 
   return lsm_error_ok;
 }

lsm/src/store/lsm_store_entry.c

@@ -63,18 +63,29 @@ void lsm_entry_close(lsm_entry_handle *handle) {
     fclose(handle->f);
   }
 
-  // TODO handle errors here
+  bool state_new = handle->states & lsm_entry_handle_state_new;
-  if ((handle->states & lsm_entry_handle_state_new) &&
+  bool state_removed = handle->states & lsm_entry_handle_state_removed;
-      !(handle->states & lsm_entry_handle_state_removed)) {
+  /* bool state_updated = handle->states & lsm_entry_handle_state_updated; */
+
+  // Clean new entry
+  if (state_new && !state_removed) {
     lsm_entry_disk_insert(handle);
-  } else if ((handle->states & lsm_entry_handle_state_removed) &&
+  }
-             !(handle->states & lsm_entry_handle_state_new)) {
+  // New entry that was removed before being written to disk; only its data file
+  // needs to be removed if present
+  else if (state_new && state_removed) {
+    lsm_entry_data_remove(handle);
+
+    lsm_entry_free(handle->wrapper->entry);
+    handle->wrapper->entry = NULL;
+  }
+  // Previously stored entry that needs to be removed; should be removed from db
+  // file as well
+  else if (state_removed && !state_new) {
     lsm_entry_disk_remove(handle);
 
     lsm_entry_free(handle->wrapper->entry);
     handle->wrapper->entry = NULL;
-  } else if (handle->states & lsm_entry_handle_state_updated) {
-    /* lsm_entry_disk_update(handle); */
   }
 
   pthread_rwlock_unlock(&handle->wrapper->lock);
@@ -316,3 +327,24 @@ lsm_error lsm_entry_data_open_read(lsm_entry_handle *handle) {
 
   return lsm_error_ok;
 }
+
+
+lsm_error lsm_entry_data_remove(lsm_entry_handle *handle) {
+  const lsm_entry *entry = handle->wrapper->entry;
+
+  if (entry->data_len > 0) {
+    if (handle->f != NULL) {
+      fclose(handle->f);
+      handle->f = NULL;
+    }
+
+    char data_path[lsm_entry_data_path_len(handle) + 1];
+    lsm_entry_data_path(data_path, handle);
+
+    if (remove(data_path) != 0) {
+      return lsm_error_failed_io;
+    }
+  }
+
+  return lsm_error_ok;
+}