incremental: fix data raw crash and improve incremental update

libzdb/data.c

@@ -325,6 +325,9 @@ data_raw_t data_raw_get_real(int fd, off_t offset) {
         return raw;
     }
 
+    // default flag raw into error mode
+    raw.error = DATA_RAW_UNEXPECTED;
+
     // moving to the header offset
     lseek(fd, offset, SEEK_SET);
 
@@ -375,6 +378,9 @@ data_raw_t data_raw_get_real(int fd, off_t offset) {
     // this validate return object to be valid
     raw.payload.length = raw.header.datalength;
 
+    // reset raw error flag, everything is fine
+    raw.error = 0;
+
     return raw;
 }
 

libzdb/data.h

@@ -76,6 +76,7 @@
 
     typedef enum data_error_t {
         DATA_RAW_EOF = 1,
+        DATA_RAW_UNEXPECTED = 2,
 
     } data_error_t;
 

tools/incremental-update/incremental.py

@@ -5,11 +5,14 @@
 
 class ZDBIncremental:
     def __init__(self, master, mport, slave, sport):
-        self.minfo = {"host": master, "port": mport}
-        self.sinfo = {"host": slave, "port": sport}
+        minfo = {"host": master, "port": mport, "namespace": "default", "name": "master"}
+        sinfo = {"host": slave, "port": sport, "namespace": "default", "name": "slave"}
 
         self.master = redis.Redis(host=master, port=mport)
+        self.master.__data = minfo
+
         self.slave = redis.Redis(host=slave, port=sport)
+        self.slave.__data = sinfo
 
         # disable defaults callbacks
         for target in [self.master, self.slave]:
@@ -19,15 +22,21 @@ def __init__(self, master, mport, slave, sport):
             target.set_response_callback("AUTH", bytes)
 
     def authenticate(self, target, password):
-        print(f"[+] authenticating")
+        print(f"[+] authenticating: {target.__data['name']}")
 
         request = target.execute_command("AUTH", "SECURE", "CHALLENGE")
         challenge = request.decode('utf-8')
 
         encoded = f"{challenge}:{password}"
         response = hashlib.sha1(encoded.encode("utf-8")).hexdigest()
 
+        # authenticate
         status = target.execute_command("AUTH", "SECURE", response)
+        if status != b"OK":
+            return False
+
+        # authenticate on namespace as well
+        status = target.execute_command("SELECT", target.__data["namespace"], password)
 
         return status == b"OK"
 
@@ -56,22 +65,20 @@ def sync(self, master, slave):
 
 
     def run(self):
-        namespace = "default"
-
-        print(f"[+] master host: {self.minfo['host']}, port: {self.minfo['port']}")
-        print(f"[+] slave host: {self.sinfo['host']}, port: {self.sinfo['port']}")
-        print(f"[+] syncing namespace: {namespace}")
+        print(f"[+] master host: {self.master.__data['host']}, port: {self.master.__data['port']}")
+        print(f"[+] slave host: {self.slave.__data['host']}, port: {self.slave.__data['port']}")
+        print(f"[+] syncing namespaces: {self.master.__data['namespace']} -> {self.slave.__data['namespace']}")
 
         while True:
             master = {}
             slave = {}
 
-            nsmaster = self.master.execute_command("NSINFO", namespace)
+            nsmaster = self.master.execute_command("NSINFO", self.master.__data['namespace'])
             master['dataid'] = int(nsmaster['data_current_id'])
             master['offset'] = int(nsmaster['data_current_offset'])
             master['size'] = int(nsmaster['data_size_bytes'])
 
-            nsslave = self.slave.execute_command("NSINFO", namespace)
+            nsslave = self.slave.execute_command("NSINFO", self.slave.__data['namespace'])
             slave['dataid'] = int(nsslave['data_current_id'])
             slave['offset'] = int(nsslave['data_current_offset'])
             slave['size'] = int(nsslave['data_size_bytes'])
@@ -99,5 +106,6 @@ def run(self):
 
 if __name__ == '__main__':
     incremental = ZDBIncremental("hub.grid.tf", 9900, "127.0.0.1", 9900)
-    # incremental.authenticate(incremental.master, "set-password-here")
+    incremental.authenticate(incremental.master, "master-password")
+    incremental.authenticate(incremental.slave, "slave-password")
     incremental.run()

zdbd/commands_system.c

@@ -275,6 +275,12 @@ static int command_data_raw(redis_client_t *client) {
         return 1;
     }
 
+    if(raw.error == DATA_RAW_UNEXPECTED) {
+        zdb_log("[-] command: data: raw: unexpected error, skipping\n");
+        redis_hardsend(client, "-Unexpected Internal Error");
+        return 1;
+    }
+
     if(raw.header.datalength == 0 || raw.header.datalength != raw.payload.length) {
         if((raw.header.flags & DATA_ENTRY_DELETED) == 0) {
             // something went wrong when fetching data from file