Merge pull request #2 from wvangeit/retry_write

Add file write retries to work around failed file downloads on osparc

src/handshakers.py

@@ -28,6 +28,7 @@ def __init__(
         self.self_uuid = str(self_uuid)
         self.other_uuid = None
 
+        self.last_write = None
         self.is_initiator = is_initiator
 
         self.input_dir_path = input_dir_path
@@ -54,6 +55,16 @@ def shake(self):
         else:
             return self.shake_receiver()
 
+    def write_filecontent(self, path: pl.Path, content: str):
+        path.write_text(content)
+
+        self.last_write = path, content
+
+    def retry_last_write(self):
+        if self.last_write is not None:
+            path, content = self.last_write
+            self.write_filecontent(path, content)
+
     def shake_initiator(self):
         """Shake hand by initiator"""
 
@@ -63,7 +74,9 @@ def shake_initiator(self):
         }
         if self.handshake_output_path.exists():
             self.handshake_output_path.unlink()
-        self.handshake_output_path.write_text(json.dumps(handshake_out))
+        self.write_filecontent(
+            self.handshake_output_path, json.dumps(handshake_out)
+        )
         logger.info(f"Wrote handshake file to {self.handshake_output_path}")
 
         def try_handshake():
@@ -89,6 +102,8 @@ def try_handshake():
                         "Waiting for correct handshake registration "
                         "confirmation ..."
                     )
+                    self.retry_last_write()
+
                 waiter += 1
             time.sleep(self.polling_interval)
 
@@ -101,7 +116,9 @@ def try_handshake():
         }
         if self.handshake_output_path.exists():
             self.handshake_output_path.unlink()
-        self.handshake_output_path.write_text(json.dumps(handshake_out))
+        self.write_filecontent(
+            self.handshake_output_path, json.dumps(handshake_out)
+        )
 
         assert other_uuid is not None
 
@@ -135,9 +152,11 @@ def shake_receiver(self):
                         "uuid": self.self_uuid,
                         "confirmed_uuid": other_uuid,
                     }
-                    self.handshake_output_path.write_text(
-                        json.dumps(handshake_out)
+
+                    self.write_filecontent(
+                        self.handshake_output_path, json.dumps(handshake_out)
                     )
+
                     logger.info(
                         f"Wrote handshake confirmation to {self.handshake_output_path}"
                     )
@@ -154,6 +173,8 @@ def shake_receiver(self):
 
             if waiter % self.print_polling_interval == 0:
                 logger.info("Waiting for registration confirmation ...")
+                self.retry_last_write()
+
             time.sleep(self.polling_interval)
             waiter += 1
 
@@ -176,5 +197,6 @@ def read_until_path_exists(self, path, wait_message="Waiting..."):
 
             if waiter % self.print_polling_interval == 0:
                 logger.debug(wait_message)
+                self.retry_last_write()
             time.sleep(self.polling_interval)
             waiter += 1

tests/functional/test_handshake_functional.py

@@ -30,5 +30,7 @@ def test_handshakes_functional(tmp_path):
         run_handshake, receiver_uuid, test_output_dir, test_input_dir, False
     )
 
+    executor.shutdown()
+
     assert initiator_other_uuid.result() == receiver_uuid
     assert receiver_other_uuid.result() == initiator_uuid

tests/unit/test_handshake.py

@@ -136,7 +136,7 @@ def mock_write_text(content):
     assert other_uuid == initiator_uuid
     assert receiver_handshake.other_uuid == initiator_uuid
 
-    assert len(written_texts) == 1
+    assert len(written_texts) > 1
 
     confirmed_handshake = {
         "uuid": receiver_uuid,