Merge pull request #148 from threefoldtech/development-v2-rscan-hotfix

libzdb: fix invalid RSCAN on sequential database

Author: maxux

README.md

@@ -188,6 +188,7 @@ This mode is not possible if you don't have any data/index already available.
 - `NSINFO <namespace>`
 - `NSLIST`
 - `NSSET <namespace> <property> <value>`
+- `NSJUMP`
 - `SELECT <namespace> [SECURE password]`
 - `DBSIZE`
 - `TIME`
@@ -367,6 +368,9 @@ denied with an error message (eg: `Namespace is temporarily locked`).
 `FREEZE` mode will deny any operation on the specific namespace, read, write, update, delete operations
 will be denied with an error message (eg: `Namespace is temporarily frozen`)
 
+## NSJUMP
+Force closing current index and data files and open the next id.
+
 ## SELECT
 Change your current namespace. If the requested namespace is password-protected, you need
 to add the password as extra parameter. If the namespace is `public` but password protected,

libzdb/index.c

@@ -25,7 +25,7 @@ void index_item_header_dump(index_item_t *item) {
     zdb_debug("[+] index: item dump: data fileid: %" PRIu32 "\n", item->dataid);
     zdb_debug("[+] index: item dump: data offset: %" PRIu32 "\n", item->offset);
     zdb_debug("[+] index: item dump: data length: %" PRIu32 "\n", item->length);
-    zdb_debug("[+] index: item dump: previous   : %" PRIx32 "\n", item->previous);
+    zdb_debug("[+] index: item dump: previous   : %" PRIu32 "\n", item->previous);
     zdb_debug("[+] index: item dump: flags      : %" PRIu8  "\n", item->flags);
     zdb_debug("[+] index: item dump: timestamp  : %" PRIu32 "\n", item->timestamp);
     zdb_debug("[+] index: item dump: parent id  : %" PRIu32 "\n", item->parentid);

libzdb/index_scan.c

@@ -60,13 +60,16 @@ void __ditry_seqmode_fix(index_item_t *source, off_t original) {
     if(zdb_rootsettings.mode != ZDB_MODE_SEQUENTIAL && zdb_rootsettings.mode != ZDB_MODE_DIRECT_KEY)
         return;
 
+    // hotfix
+    return;
+
     // compute previous offset
-    // since sequential keys are hardcoded to be 4 bytes, we can easily
+    // since sequential keys are hardcoded, we can easily
     // compute previous offset and ignoring value in the file
     //
     // this value were incorrect in some early version
     //
-    source->previous = original - sizeof(index_item_t) - sizeof(uint32_t);
+    source->previous = original - sizeof(index_item_t) - sizeof(seqid_t);
 
     if(source->previous == sizeof(index_header_t))
         source->previous = 1;
@@ -76,12 +79,14 @@ void __ditry_seqmode_fix(index_item_t *source, off_t original) {
 static index_scan_t index_previous_header_real(index_scan_t scan) {
     index_item_t source;
 
+    #if 0
     // special dirty-fix case, please see __ditry_seqmode_fix function
     if(scan.target == 1) {
         // forcing last entry position
         scan.target = lseek(scan.fd, 0, SEEK_END);
         scan.target -= sizeof(index_item_t) + sizeof(uint32_t);
     }
+    #endif
 
     // if scan.target is not set yet, we don't know the expected
     // offset of the previous header, let's read the header
@@ -94,11 +99,12 @@ static index_scan_t index_previous_header_real(index_scan_t scan) {
             return index_scan_error(scan, INDEX_SCAN_UNEXPECTED);
         }
 
-        __ditry_seqmode_fix(&source, scan.original);
+        // __ditry_seqmode_fix(&source, scan.original);
         index_item_header_dump(&source);
 
         if(source.previous >= current) {
             zdb_debug("[+] index rscan: previous-header: previous offset (%u) in previous file\n", source.previous);
+            scan.target = source.previous;
             return index_scan_error(scan, INDEX_SCAN_REQUEST_PREVIOUS);
         }
 
@@ -117,11 +123,13 @@ static index_scan_t index_previous_header_real(index_scan_t scan) {
         return index_scan_error(scan, INDEX_SCAN_NO_MORE_DATA);
     }
 
+    #if 0
     // special dirty-fix case, please see __ditry_seqmode_fix function
     if(scan.target == 1) {
         scan.target = 0;
         return index_scan_error(scan, INDEX_SCAN_REQUEST_PREVIOUS);
     }
+    #endif
 
     // jumping to previous object
     lseek(scan.fd, scan.target, SEEK_SET);
@@ -132,7 +140,7 @@ static index_scan_t index_previous_header_real(index_scan_t scan) {
         return index_scan_error(scan, INDEX_SCAN_UNEXPECTED);
     }
 
-    __ditry_seqmode_fix(&source, scan.target);
+    // __ditry_seqmode_fix(&source, scan.target);
 
     // checking if entry is deleted
     if(source.flags & INDEX_ENTRY_DELETED) {

zdbd/commands.c

@@ -151,6 +151,7 @@ static command_t commands_handlers[] = {
     {.command = "NSLIST",  .handler = command_nslist},   // custom command to list namespaces
     {.command = "NSSET",   .handler = command_nsset},    // custom command to edit namespace settings
     {.command = "NSINFO",  .handler = command_nsinfo},   // custom command to get namespace information
+    {.command = "NSJUMP",  .handler = command_nsjump},   // custom command to force jumping to next index/data
     {.command = "SELECT",  .handler = command_select},   // default SELECT (with pwd) namespace switch
     {.command = "RELOAD",  .handler = command_reload},   // custom command to reload a namespace
     {.command = "FLUSH",   .handler = command_flush},    // custom command to reset a namespace

zdbd/commands_namespace.c

@@ -582,6 +582,21 @@ int command_nsset(redis_client_t *client) {
     return 0;
 }
 
+int command_nsjump(redis_client_t *client) {
+    // command restricted to admin only
+    if(!command_admin_authorized(client))
+        return 1;
+
+    zdbd_debug("[+] command: nsjump: forcing index and data jump\n");
+
+    size_t newid = index_jump_next(client->ns->index);
+    data_jump_next(client->ns->data, newid);
+
+    redis_hardsend(client, "+OK");
+
+    return 0;
+}
+
 int command_dbsize(redis_client_t *client) {
     char response[64];
 

zdbd/commands_namespace.h

@@ -7,6 +7,7 @@
     int command_nslist(redis_client_t *client);
     int command_nsinfo(redis_client_t *client);
     int command_nsset(redis_client_t *client);
+    int command_nsjump(redis_client_t *client);
     int command_dbsize(redis_client_t *client);
     int command_reload(redis_client_t *client);
     int command_flush(redis_client_t *client);