Retry GCP metadata writes on an operation-level fingerprint conflict

CHANGELOG.rst

@@ -1,3 +1,28 @@
+4.4.2 - unreleased
+------------------
+
+## Fixes
+* **GCP common-metadata writes now survive concurrent writers.** Labels for
+  networks, routers, firewalls and key pairs live in the project-wide common
+  instance metadata, which every write re-uploads under an optimistic
+  fingerprint. Two things went wrong under parallel use. The retry for a
+  stale fingerprint only recognised the conflict as an HTTP error, but a
+  concurrent writer produces it differently: the upload is accepted and the
+  resulting *operation* completes with ``CONDITION_NOT_MET``, which
+  ``wait_for_operation`` raised as a plain ``Exception`` the retry ignored.
+  And - measured in the live suite - GCP accepts several writes against the
+  same fingerprint while an earlier one is still pending, and a later one can
+  complete as ``DONE`` with its change absent from the document: 11 of 56
+  writes in one run. ``wait_for_operation`` now raises ``GCPOperationError``
+  (a ``ProviderInternalException`` carrying the operation's error payload and
+  ``codes``); a metadata write is only finished once its change reads back,
+  and is redone on fresh metadata otherwise (``MetadataWriteNotApplied`` once
+  the retries are exhausted); a write that changes nothing is not sent at
+  all; ``add_metadata_item`` checks for an existing key itself and raises
+  ``DuplicateResourceException``, so a retry never appends a second copy;
+  and ``remove_metadata_item`` returns ``False`` when there was nothing to
+  remove, as its callers always assumed.
+
 4.4.1 - August 21, 2026 (sha 093ef669598d9f324be28d400a851396739cf1d8)
 ----------------------------------------------------------------------
 

cloudbridge/providers/gcp/helpers.py

@@ -13,12 +13,45 @@
 
 import tenacity
 
+from cloudbridge.interfaces.exceptions import DuplicateResourceException
 from cloudbridge.interfaces.exceptions import ProviderInternalException
 
 if TYPE_CHECKING:
     from .provider import GCPCloudProvider
 
 
+class GCPOperationError(ProviderInternalException):
+    """A GCP operation completed with an error.
+
+    Raised by ``wait_for_operation`` with the operation's ``error`` payload,
+    so callers can act on the error codes rather than parse the message.
+    """
+
+    def __init__(self, error: Any) -> None:
+        super().__init__(error)
+        self.error = error
+
+    @property
+    def codes(self) -> list[str]:
+        return [e.get('code', '') for e in self.error.get('errors', [])]
+
+
+class MetadataWriteNotApplied(ProviderInternalException):
+    """A common-metadata write completed as DONE but is absent when read back.
+
+    Observed under concurrent writers: GCP accepts several writes against
+    the same fingerprint while an earlier one is still pending, and a later
+    one can complete without its change surviving. The write is retried on
+    fresh metadata; this is raised only once the retries are exhausted.
+    """
+
+    def __init__(self, keys: list[str]) -> None:
+        super().__init__(
+            'Project metadata write reported done, but these keys do not '
+            'hold the written value: {}'.format(', '.join(keys)))
+        self.keys = keys
+
+
 def gcp_projects(provider: "GCPCloudProvider") -> Any:
     return provider.gcp_compute.projects()
 
@@ -43,8 +76,18 @@ def get_common_metadata(provider: "GCPCloudProvider") -> Any:
     return metadata["commonInstanceMetadata"]
 
 
-def __if_fingerprint_differs(e: BaseException) -> bool:
-    # return True if the CloudError exception is due to subnet being in use
+def __metadata_write_lost_to_a_concurrent_writer(e: BaseException) -> bool:
+    """Whether ``e`` means a metadata write must be redone on fresh metadata.
+
+    A concurrent writer shows up in three shapes: an HTTP 412 on the request
+    itself, a submitted operation that completes with ``CONDITION_NOT_MET``,
+    or - when GCP accepted both writes against the same fingerprint - an
+    operation that completes as DONE without the change surviving.
+    """
+    if isinstance(e, GCPOperationError):
+        return 'CONDITION_NOT_MET' in e.codes
+    if isinstance(e, MetadataWriteNotApplied):
+        return True
     if isinstance(e, HttpError):
         expected_message = 'Supplied fingerprint does not match current ' \
                            'metadata fingerprint.'
@@ -53,8 +96,14 @@ def __if_fingerprint_differs(e: BaseException) -> bool:
     return False
 
 
+def _metadata_values(metadata: Any) -> dict[str, Any]:
+    return {item['key']: item.get('value')
+            for item in metadata.get('items', [])}
+
+
 @tenacity.retry(stop=tenacity.stop_after_attempt(10),
-                retry=tenacity.retry_if_exception(__if_fingerprint_differs),
+                retry=tenacity.retry_if_exception(
+                    __metadata_write_lost_to_a_concurrent_writer),
                 wait=tenacity.wait_exponential(max=10),
                 reraise=True)
 def gcp_metadata_save_op(provider: "GCPCloudProvider",
@@ -66,16 +115,36 @@ def gcp_metadata_save_op(provider: "GCPCloudProvider",
     retrieves the metadata, invokes the provided callback with that
     metadata, and saves the metadata using the original fingerprint
     immediately afterwards, ensuring that update conflicts can be detected.
+
+    The fingerprint alone is not enough: a write that GCP reports as done
+    can still be missing from the document when another write was accepted
+    against the same fingerprint. So the write is only finished once the
+    callback's changes can be read back, and is redone on fresh metadata
+    otherwise. A callback that changes nothing sends no write at all - every
+    write re-uploads the whole document and moves the fingerprint, which
+    would only add contention for other writers.
     """
     def _save_common_metadata(provider: "GCPCloudProvider") -> None:
         # get the latest metadata (so we get the latest fingerprint)
         metadata = get_common_metadata(provider)
+        before = _metadata_values(metadata)
         # allow callback to do processing on it
         callback(metadata)
+        wanted = _metadata_values(metadata)
+        changes = {key: wanted.get(key) for key in before.keys() | wanted.keys()
+                   if before.get(key) != wanted.get(key)}
+        if not changes:
+            return
         # save the metadata
         operation = gcp_projects(provider).setCommonInstanceMetadata(
             project=provider.project_name, body=metadata).execute()
         provider.wait_for_operation(operation)
+        # ...and make sure it stuck
+        current = _metadata_values(get_common_metadata(provider))
+        lost = sorted(key for key, value in changes.items()
+                      if current.get(key) != value)
+        if lost:
+            raise MetadataWriteNotApplied(lost)
 
     # Retry a few times if the fingerprints conflict
     _save_common_metadata(provider)
@@ -98,15 +167,24 @@ def _update_metadata_key(metadata: Any) -> None:
     gcp_metadata_save_op(provider, _update_metadata_key)
 
 
-# This function will raise an HttpError with message containing
-# "Metadata has duplicate key" if it's not unique, unlike the previous
-# method which either adds or updates the value corresponding to that key
 def add_metadata_item(provider: "GCPCloudProvider", key: str,
                       value: str) -> None:
+    """Add ``key``, raising ``DuplicateResourceException`` if it is taken.
+
+    Unlike ``modify_or_add_metadata_item`` this never overwrites. The
+    check is done on the freshly fetched metadata so that a retry after a
+    lost write does not append a second copy of a key that did land: a key
+    already holding this very value is this write, and counts as done.
+    """
     def _add_metadata_key(metadata: Any) -> None:
-        entry = {'key': key, 'value': value}
         entries = metadata.get('items', [])
-        entries.append(entry)
+        existing = [item for item in entries if item['key'] == key]
+        if existing:
+            if existing[-1].get('value') == value:
+                return
+            raise DuplicateResourceException(
+                'Metadata key {0} already exists'.format(key))
+        entries.append({'key': key, 'value': value})
         # Reassign explicitly in case the original get returned [] although
         # if not it will be already updated
         metadata['items'] = entries
@@ -136,30 +214,25 @@ def get_metadata_item_value(provider: "GCPCloudProvider", key: str) -> Any:
 
 
 def remove_metadata_item(provider: "GCPCloudProvider", key: str) -> bool:
-    def _remove_metadata_by_key(metadata: Any) -> bool | None:
-        items = metadata.get('items', [])
-        # No metadata to delete
-        if not items:
-            return False
-        else:
-            entries = [item for item in metadata.get('items', [])
-                       if item['key'] != key]
-
-            # Make sure only one entry is deleted
-            if len(entries) < len(items) - 1:
-                raise ProviderInternalException("Multiple metadata entries "
-                                                "found for the same key {}"
-                                                .format(key))
-            # If none is deleted indicate so by returning False
-            elif len(entries) == len(items):
-                return False
+    """Remove ``key``; ``False`` if there was nothing to remove."""
+    removed = False
 
-            else:
-                metadata['items'] = entries
-                return None
+    def _remove_metadata_by_key(metadata: Any) -> None:
+        nonlocal removed
+        items = metadata.get('items', [])
+        entries = [item for item in items if item['key'] != key]
+        # Make sure only one entry is deleted
+        if len(entries) < len(items) - 1:
+            raise ProviderInternalException("Multiple metadata entries "
+                                            "found for the same key {}"
+                                            .format(key))
+        # A retry after a lost write finds the key still present and
+        # removes it again; the first attempt decides whether it was there.
+        removed = removed or len(entries) < len(items)
+        metadata['items'] = entries
 
     gcp_metadata_save_op(provider, _remove_metadata_by_key)
-    return True
+    return removed
 
 
 def __if_label_fingerprint_differs(e: BaseException) -> bool:

cloudbridge/providers/gcp/provider.py

@@ -30,6 +30,7 @@
 from cloudbridge.interfaces.services import SecurityService
 from cloudbridge.interfaces.services import StorageService
 
+from .helpers import GCPOperationError
 from .services import GCPComputeService
 from .services import GCPDnsService
 from .services import GCPNetworkingService
@@ -401,7 +402,7 @@ def wait_for_operation(self, operation: Any, region: str | None = None,
             result = operations.get(**args).execute()
             if result['status'] == 'DONE':
                 if 'error' in result:
-                    raise Exception(result['error'])
+                    raise GCPOperationError(result['error'])
                 return result
 
             time.sleep(0.5)

cloudbridge/providers/gcp/services.py

@@ -188,7 +188,12 @@ def create(self, name: str,
                                       GCPKeyPair.KP_TAG_PREFIX + name,
                                       metadata_value)
             return GCPKeyPair(provider, kp_info, private_key)
+        except DuplicateResourceException:
+            raise DuplicateResourceException(
+                'A KeyPair with name {0} already exists'.format(name))
         except googleapiclient.errors.HttpError as err:
+            # GCP's own duplicate-key rejection, should a concurrent create
+            # slip past the check on the fetched metadata.
             if err.resp.get('content-type', '').startswith('application/json'):
                 message = (json.loads(err.content).get('error', {})
                            .get('errors', [{}])[0].get('message'))