feat(lsm): implement updating entries

storage

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased](https://git.rustybever.be/Chewing_Bever/lander/src/branch/dev)
 
+## Fixed
+
+* Failed uploads now no longer leave behind a partial entry file
+
 ## [0.2.1](https://git.rustybever.be/Chewing_Bever/lander/src/tag/0.2.1)
 
 ### Added

include/lander.h

@@ -51,6 +51,8 @@ lnm_http_step_err lander_post_paste_secure(lnm_http_conn *conn);
 
 lnm_http_step_err lander_stream_body_to_entry(lnm_http_conn *conn);
 
+lnm_http_step_err lander_commit_entry(lnm_http_conn *conn);
+
 lnm_http_step_err lander_post_redirect_body_to_attr(lnm_http_conn *conn);
 
 lnm_http_step_err lander_remove_entry(lnm_http_conn *conn);

lsm/include/lsm/store.h

@@ -159,6 +159,15 @@ lsm_error lsm_store_open_read(lsm_entry_handle **out, lsm_store *store,
 lsm_error lsm_store_open_write(lsm_entry_handle **out, lsm_store *store,
                                const lsm_str *key);
 
+/**
+ * Commit any changes to the persistent storage. Any changes, insertions or
+ * deletions that occured without a commit are reverted when the handle is
+ * closed.
+ *
+ * @param handle handle to the entry
+ */
+lsm_error lsm_entry_commit(lsm_entry_handle *handle);
+
 /**
  * Close an open entry handle.
  *

lsm/src/_include/lsm/store_internal.h

@@ -157,7 +157,8 @@ 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.
+ * Remove the entry's data file if present and close its handle. Any uncommitted
+ * changes will be reverted.
  *
  * @param handle handle to the entry
  */

lsm/src/store/lsm_store_disk_write.c

@@ -1,3 +1,5 @@
+#include <stdint.h>
+
 #include "lsm/store_internal.h"
 
 static lsm_error lsm_fwrite(uint64_t *sum, FILE *f, uint64_t size,
@@ -125,16 +127,10 @@ lsm_error lsm_entry_disk_insert(lsm_entry_handle *handle) {
   return res;
 }
 
-// Marking an entry as removed in the idx file is simply setting the length of
+static lsm_error lsm_idx_zero_block(lsm_store *store, uint64_t pos) {
-// its entry to zero
-lsm_error lsm_entry_disk_remove(lsm_entry_handle *handle) {
-  lsm_store *store = handle->store;
-  const lsm_entry *entry = handle->wrapper->entry;
-
   pthread_mutex_lock(&store->idx.lock);
 
-  lsm_error res =
+  lsm_error res = lsm_fseek(store->idx.f, pos);
-      lsm_fseek(store->idx.f, entry->idx_file_offset + sizeof(uint64_t));
 
   if (res != lsm_error_ok) {
     pthread_mutex_unlock(&store->idx.lock);
@@ -153,7 +149,29 @@ lsm_error lsm_entry_disk_remove(lsm_entry_handle *handle) {
 
   fflush(store->idx.f);
 
+  return lsm_error_ok;
+}
+
+// Marking an entry as removed in the idx file is simply setting the length of
+// its entry to zero
+lsm_error lsm_entry_disk_remove(lsm_entry_handle *handle) {
+  const lsm_entry *entry = handle->wrapper->entry;
+
+  LSM_RES(lsm_idx_zero_block(handle->store,
+                             entry->idx_file_offset * sizeof(uint64_t)));
   LSM_RES(lsm_entry_data_remove(handle));
 
   return lsm_error_ok;
 }
+
+lsm_error lsm_entry_disk_update(lsm_entry_handle *handle) {
+  // An update is implemented by reinserting the entry at the end of the db file
+  uint64_t old_idx_index = handle->wrapper->entry->idx_file_offset;
+
+  // TODO is there any way we can make this atomic? If the zero write to the
+  // index file fails, there are two entries in the db file for the same key.
+  LSM_RES(lsm_entry_disk_insert(handle));
+  LSM_RES(lsm_idx_zero_block(handle->store, old_idx_index * sizeof(uint64_t)));
+
+  return lsm_error_ok;
+}

lsm/src/store/lsm_store_entry.c

@@ -58,35 +58,51 @@ lsm_error lsm_entry_handle_init(lsm_entry_handle **out) {
   return lsm_error_ok;
 }
 
-void lsm_entry_close(lsm_entry_handle *handle) {
+lsm_error lsm_entry_commit(lsm_entry_handle *handle) {
-  if (handle->f != NULL) {
+  uint8_t state_new = handle->states & lsm_entry_handle_state_new;
-    fclose(handle->f);
+  uint8_t state_removed = handle->states & lsm_entry_handle_state_removed;
-  }
+  uint8_t state_updated = handle->states & lsm_entry_handle_state_updated;
-
-  bool state_new = handle->states & lsm_entry_handle_state_new;
-  bool 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);
+    LSM_RES(lsm_entry_disk_insert(handle));
   }
-  // New entry that was removed before being written to disk; only its data file
+  // Previously stored entry that needs to be removed; should be removed from db
-  // needs to be removed if present
+  // file as well
-  else if (state_new && state_removed) {
+  else if (state_removed && !state_new) {
+    LSM_RES(lsm_entry_disk_remove(handle));
+
+    lsm_entry_free(handle->wrapper->entry);
+    handle->wrapper->entry = NULL;
+  } else if (state_updated && !(state_new || state_removed)) {
+    LSM_RES(lsm_entry_disk_update(handle));
+  }
+
+  // Reset states after committing current changes
+  handle->states = 0;
+
+  return lsm_error_ok;
+}
+
+void lsm_entry_close(lsm_entry_handle *handle) {
+  if (handle->f != NULL) {
+    fclose(handle->f);
+    handle->f = NULL;
+  }
+
+  uint8_t state_new = handle->states & lsm_entry_handle_state_new;
+  /* bool state_updated = handle->states & lsm_entry_handle_state_updated; */
+
+  // New entries create a wrapper in the trie that should be removed if not
+  // committed
+  if (state_new) {
     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);
+  // TODO rollback uncomitted updates
-    handle->wrapper->entry = NULL;
-  }
 
   pthread_rwlock_unlock(&handle->wrapper->lock);
   free(handle);
@@ -328,7 +344,6 @@ 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;
 

src/lander/lander_steps.c

@@ -1,6 +1,7 @@
 #include <string.h>
 
 #include "lnm/http/loop.h"
+#include "lnm/log.h"
 #include "lnm/loop.h"
 
 #include "lander.h"
@@ -25,3 +26,12 @@ lnm_http_step_err lander_stream_body_to_entry(lnm_http_conn *conn) {
              ? lnm_http_step_err_done
              : lnm_http_step_err_io_needed;
 }
+
+lnm_http_step_err lander_commit_entry(lnm_http_conn *conn) {
+  lnm_http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+
+  lsm_entry_commit(c_ctx->entry);
+
+  return lnm_http_step_err_done;
+}

src/main.c

@@ -3,6 +3,7 @@
 #include <time.h>
 
 #include "lnm/http/loop.h"
+#include "lnm/http/route.h"
 #include "lnm/log.h"
 
 #include "lander.h"
@@ -30,54 +31,64 @@ lnm_http_loop *loop_init(lander_gctx *gctx, const char *api_key) {
   lnm_http_router_add(&route, router, lnm_http_method_delete, "/:key");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_remove_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/s/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_redirect, false);
   lnm_http_route_step_append(route, lnm_http_loop_step_body_to_buf, false);
   lnm_http_route_step_append(route, lander_post_redirect_body_to_attr, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/sl/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_redirect_secure, false);
   lnm_http_route_step_append(route, lnm_http_loop_step_body_to_buf, false);
   lnm_http_route_step_append(route, lander_post_redirect_body_to_attr, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/s/:key");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_redirect, false);
   lnm_http_route_step_append(route, lnm_http_loop_step_body_to_buf, false);
   lnm_http_route_step_append(route, lander_post_redirect_body_to_attr, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/p/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_paste, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/pl/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_paste_secure, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/p/:key");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_paste, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/f/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_file, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/fl/");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_file_secure, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_router_add(&route, router, lnm_http_method_post, "/f/:key");
   lnm_http_route_step_append(route, lnm_http_loop_step_auth, false);
   lnm_http_route_step_append(route, lander_post_file, false);
   lnm_http_route_step_append(route, lander_stream_body_to_entry, false);
+  lnm_http_route_step_append(route, lander_commit_entry, true);
 
   lnm_http_loop_router_set(hl, router);