Allow payment reuse for same app LRN (#961)

Part of [Service provider auctions for web deployments](https://www.notion.so/Service-provider-auctions-for-web-deployments-104a6b22d47280dbad51d28aa3a91d75)

Reviewed-on: #961
Reviewed-by: ashwin <ashwin@noreply.git.vdb.to>
Co-authored-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
Co-committed-by: Prathamesh Musale <prathamesh.musale0@gmail.com>

stack_orchestrator/deploy/webapp/registry_mutex.py

@@ -1,8 +1,59 @@
-import fcntl
 from functools import wraps
+import os
+import time
 
 # Define default file path for the lock
 DEFAULT_LOCK_FILE_PATH = "/tmp/registry_mutex_lock_file"
+LOCK_TIMEOUT = 30
+LOCK_RETRY_INTERVAL = 3
+
+
+def acquire_lock(client, lock_file_path, timeout):
+    # Lock alreay acquired by the current client
+    if client.mutex_lock_acquired:
+        return
+
+    while True:
+        try:
+            # Check if lock file exists and is potentially stale
+            if os.path.exists(lock_file_path):
+                with open(lock_file_path, 'r') as lock_file:
+                    timestamp = float(lock_file.read().strip())
+
+                # If lock is stale, remove the lock file
+                if time.time() - timestamp > timeout:
+                    print(f"Stale lock detected, removing lock file {lock_file_path}")
+                    os.remove(lock_file_path)
+                else:
+                    print(f"Lock file {lock_file_path} exists and is recent, waiting...")
+                    time.sleep(LOCK_RETRY_INTERVAL)
+                    continue
+
+            # Try to create a new lock file with the current timestamp
+            fd = os.open(lock_file_path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
+            with os.fdopen(fd, 'w') as lock_file:
+                lock_file.write(str(time.time()))
+
+            client.mutex_lock_acquired = True
+            print(f"Registry lock acquired, {lock_file_path}")
+
+            # Lock successfully acquired
+            return
+
+        except FileExistsError:
+            print(f"Lock file {lock_file_path} exists, waiting...")
+            time.sleep(LOCK_RETRY_INTERVAL)
+
+
+def release_lock(client, lock_file_path):
+    try:
+        os.remove(lock_file_path)
+
+        client.mutex_lock_acquired = False
+        print(f"Registry lock released, {lock_file_path}")
+    except FileNotFoundError:
+        # Lock file already removed
+        pass
 
 
 def registry_mutex():
@@ -10,21 +61,16 @@ def registry_mutex():
         @wraps(func)
         def wrapper(self, *args, **kwargs):
             lock_file_path = DEFAULT_LOCK_FILE_PATH
-            if self.mutex_lock_file is not None:
+            if self.mutex_lock_file:
                 lock_file_path = self.mutex_lock_file
 
-            with open(lock_file_path, 'w') as lock_file:
+            # Acquire the lock before running the function
+            acquire_lock(self, lock_file_path, LOCK_TIMEOUT)
             try:
-                    # Try to acquire the lock
+                return func(self, *args, **kwargs)
-                    fcntl.flock(lock_file, fcntl.LOCK_EX)
-
-                    # Call the actual function
-                    result = func(self, *args, **kwargs)
             finally:
-                    # Always release the lock
+                # Release the lock after the function completes
-                    fcntl.flock(lock_file, fcntl.LOCK_UN)
+                release_lock(self, lock_file_path)
-
-            return result
 
         return wrapper
 

stack_orchestrator/deploy/webapp/util.py

@@ -114,7 +114,7 @@ def is_id(name_or_id: str):
 
 
 class LaconicRegistryClient:
-    def __init__(self, config_file, log_file=None):
+    def __init__(self, config_file, log_file=None, mutex_lock_file=None):
         self.config_file = config_file
         self.log_file = log_file
         self.cache = AttrDict(
@@ -125,6 +125,9 @@ class LaconicRegistryClient:
             }
         )
 
+        self.mutex_lock_file = mutex_lock_file
+        self.mutex_lock_acquired = False
+
     def whoami(self, refresh=False):
         if not refresh and "whoami" in self.cache:
             return self.cache["whoami"]
@@ -843,16 +846,21 @@ def confirm_payment(laconic: LaconicRegistryClient, record, payment_address, min
         )
         return False
 
-    # Check if the payment was already used on a
+    # Check if the payment was already used on a deployment
     used = laconic.app_deployments(
-        {"deployer": payment_address, "payment": tx.hash}, all=True
+        {"deployer": record.attributes.deployer, "payment": tx.hash}, all=True
     )
     if len(used):
-        logger.log(f"{record.id}: payment {tx.hash} already used on deployment {used}")
+        # Fetch the app name from request record
+        used_request = laconic.get_record(used[0].attributes.request, require=True)
+
+        # Check that payment was used for deployment of same application
+        if record.attributes.application != used_request.attributes.application:
+            logger.log(f"{record.id}: payment {tx.hash} already used on a different application deployment {used}")
             return False
 
     used = laconic.app_deployment_removals(
-        {"deployer": payment_address, "payment": tx.hash}, all=True
+        {"deployer": record.attributes.deployer, "payment": tx.hash}, all=True
     )
     if len(used):
         logger.log(