Finish a GCP metadata write only once its change reads back

Retrying on CONDITION_NOT_MET turned out not to be enough. Instrumenting
the live suite showed why: with five workers labelling resources in one
project, GCP accepts several setCommonInstanceMetadata requests against
the same fingerprint while an earlier one is still pending (one
fingerprint was accepted eleven times), and a later one can complete as
DONE with its change absent from the document - 11 of 56 writes in a
single run, none of them turning up later. The pending window was wide
because the document had grown to ~1,700 orphaned test entries, making
each write take 10-260 seconds, but the behaviour is GCP's and any
concurrent writers can hit it.

So a write now counts as done only when the callback's changes can be
read back, and is redone on fresh metadata otherwise; after the retries
are exhausted it raises MetadataWriteNotApplied. A callback that changes
nothing sends no write, since every write re-uploads the whole document
and moves the fingerprint for everyone else. add_metadata_item checks the
fetched metadata for the key itself and raises DuplicateResourceException
- a key already holding this write's value counts as done, so a retry
after a lost write never appends a second copy - with GCP's own
duplicate-key rejection kept as the backstop. remove_metadata_item
returns False when there was nothing to remove, which its callers had
always tested for; it returned True unconditionally.

The tests drive the real save operation and wait_for_operation against a
fake compute client holding a server-side document whose operations can
apply, conflict, or complete without applying.

CHANGELOG.rst

@@ -2,19 +2,26 @@
 ------------------
 
 ## Fixes
-* **GCP metadata writes now retry when a concurrent writer invalidates the
-  fingerprint.** 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. 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. So under parallel use every
-  collision failed on the first attempt - the cause of the recurring
-  ``test_crud_*`` failures in the GCP live suite. ``wait_for_operation`` now
-  raises ``GCPOperationError`` (a ``ProviderInternalException`` carrying the
-  operation's error payload and ``codes``), and the metadata save retries on
-  it with freshly fetched metadata, as it always did for the HTTP form.
+* **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,6 +13,7 @@ from googleapiclient.errors import HttpError
 
 import tenacity
 
+from cloudbridge.interfaces.exceptions import DuplicateResourceException
 from cloudbridge.interfaces.exceptions import ProviderInternalException
 
 if TYPE_CHECKING:
@@ -35,6 +36,22 @@ class GCPOperationError(ProviderInternalException):
         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()
 
@@ -59,15 +76,18 @@ def get_common_metadata(provider: "GCPCloudProvider") -> Any:
     return metadata["commonInstanceMetadata"]
 
 
-def __if_fingerprint_differs(e: BaseException) -> bool:
-    """Whether ``e`` is GCP rejecting a metadata write on a stale fingerprint.
+def __metadata_write_lost_to_a_concurrent_writer(e: BaseException) -> bool:
+    """Whether ``e`` means a metadata write must be redone on fresh metadata.
 
-    The conflict surfaces in two shapes: an HTTP error on the request itself,
-    or a successfully submitted operation that then completes with
-    ``CONDITION_NOT_MET`` - which is what a concurrent writer produces.
+    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.'
@@ -76,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",
@@ -89,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)
@@ -121,15 +167,24 @@ def modify_or_add_metadata_item(provider: "GCPCloudProvider", key: str,
     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
@@ -159,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/services.py

@@ -188,7 +188,12 @@ class GCPKeyPairService(BaseKeyPairService):
                                       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'))

tests/test_gcp_metadata_save.py

@@ -1,11 +1,14 @@
-"""Retrying GCP common-metadata writes on a fingerprint conflict.
+"""Making GCP common-metadata writes stick.
 
 GCP keeps labels and key pairs in the project-wide common instance metadata,
-which every write re-uploads under an optimistic fingerprint. A concurrent
-writer makes the upload's *operation* fail with ``CONDITION_NOT_MET``; that
-is a different path from an HTTP-level error, and the write has to be retried
-with freshly fetched metadata on either. No SDK is involved: the compute
-client is a fake and the provider's real ``wait_for_operation`` polls it.
+which every write re-uploads under an optimistic fingerprint. Under
+concurrent writers two things happen that a write has to survive: the
+operation fails with ``CONDITION_NOT_MET`` (a different path from the
+HTTP-level 412), and - observed in the live suite - the operation reports
+``DONE`` yet the change is absent from the document afterwards. Both must
+lead to the write being re-applied on freshly fetched metadata. No SDK is
+involved: the compute client is a fake with a server-side document, and the
+provider's real ``wait_for_operation`` polls it.
 """
 
 import unittest
@@ -13,8 +16,13 @@ from unittest import mock
 
 import tenacity
 
+from cloudbridge.interfaces.exceptions import DuplicateResourceException
 from cloudbridge.providers.gcp.helpers import GCPOperationError
+from cloudbridge.providers.gcp.helpers import MetadataWriteNotApplied
+from cloudbridge.providers.gcp.helpers import add_metadata_item
 from cloudbridge.providers.gcp.helpers import gcp_metadata_save_op
+from cloudbridge.providers.gcp.helpers import modify_or_add_metadata_item
+from cloudbridge.providers.gcp.helpers import remove_metadata_item
 from cloudbridge.providers.gcp.provider import GCPCloudProvider
 
 FINGERPRINT_CONFLICT = {
@@ -25,6 +33,11 @@ OTHER_FAILURE = {
     'errors': [{'code': 'RESOURCE_NOT_FOUND',
                 'message': "The resource 'projects/p' was not found"}]}
 
+# Scripted outcomes for successive set operations.
+APPLY = 'apply'        # the write lands and the fingerprint moves on
+LOST = 'lost'          # reported DONE, but nothing changed
+CONFLICT = 'conflict'  # CONDITION_NOT_MET, nothing changed
+
 
 class _Call:
     def __init__(self, result):
@@ -35,85 +48,161 @@ class _Call:
 
 
 class _FakeCompute:
-    """Enough of the compute client for a metadata save: each save yields an
-    operation whose outcome is the next entry in ``operation_results``."""
-
-    def __init__(self, operation_results):
-        self.operation_results = list(operation_results)
+    """A compute client holding one project metadata document.
+
+    ``projects().get()`` returns a copy of the document; each
+    ``setCommonInstanceMetadata()`` yields an operation whose fate is the
+    next entry in ``outcomes``, resolved when ``wait_for_operation`` polls
+    ``globalOperations().get()``.
+    """
+
+    def __init__(self, outcomes, items=None):
+        self.outcomes = list(outcomes)
+        self.document = {'fingerprint': 'fp-0',
+                         'items': list(items or [])}
+        self.version = 0
         self.fetches = 0
         self.saved_bodies = []
+        self._pending = {}
 
-    # projects().get() / projects().setCommonInstanceMetadata()
     def projects(self):
         return self
 
-    def get(self, project):
+    def globalOperations(self):
+        return self
+
+    def get(self, project, operation=None):
+        if operation is not None:
+            return self._resolve(operation)
         self.fetches += 1
         return _Call({'commonInstanceMetadata': {
-            'fingerprint': f'fp-{self.fetches}', 'items': []}})
+            'fingerprint': self.document['fingerprint'],
+            'items': [dict(i) for i in self.document['items']]}})
 
     def setCommonInstanceMetadata(self, project, body):
         self.saved_bodies.append(body)
-        return _Call({'name': f'op-{len(self.saved_bodies)}'})
-
-    # globalOperations().get() - polled by wait_for_operation
-    def globalOperations(self):
-        return self
+        name = 'op-%d' % len(self.saved_bodies)
+        self._pending[name] = body
+        return _Call({'name': name})
 
-    def get_operation(self, project, operation):
-        outcome = self.operation_results.pop(0)
+    def _resolve(self, name):
+        body = self._pending.pop(name)
+        outcome = self.outcomes.pop(0)
         result = {'status': 'DONE'}
-        if outcome is not None:
+        if outcome == APPLY:
+            self.version += 1
+            self.document = {'fingerprint': 'fp-%d' % self.version,
+                             'items': [dict(i) for i in body['items']]}
+        elif outcome == CONFLICT:
+            result['error'] = FINGERPRINT_CONFLICT
+        elif outcome == LOST:
+            pass
+        else:
             result['error'] = outcome
         return _Call(result)
 
+    def keys(self):
+        return {i['key']: i['value'] for i in self.document['items']}
+
 
 class _FakeProvider:
     project_name = 'p'
     wait_for_operation = GCPCloudProvider.wait_for_operation
 
-    def __init__(self, operation_results):
-        self.gcp_compute = _FakeCompute(operation_results)
-        # wait_for_operation calls operations.get(**args); the fake's
-        # get() is taken by projects().get(project=), so route it.
-        self.gcp_compute.get = self._route_get
-
-    def _route_get(self, **kwargs):
-        if 'operation' in kwargs:
-            return self.gcp_compute.get_operation(**kwargs)
-        return _FakeCompute.get(self.gcp_compute, **kwargs)
-
-
-def _save(provider, callback):
-    # The production wait between attempts is exponential backoff; the test
-    # is about whether a retry happens, not how long it waits.
-    return gcp_metadata_save_op.retry_with(
-        wait=tenacity.wait_none())(provider, callback)
+    def __init__(self, outcomes, items=None):
+        self.gcp_compute = _FakeCompute(outcomes, items)
 
 
 class GCPMetadataSaveTestCase(unittest.TestCase):
 
+    def setUp(self):
+        # The production wait between attempts is exponential backoff; these
+        # tests are about whether a retry happens, not how long it waits.
+        patcher = mock.patch.object(gcp_metadata_save_op.retry, 'wait',
+                                    tenacity.wait_none())
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
     def test_fingerprint_conflict_is_retried_with_fresh_metadata(self):
-        provider = _FakeProvider([FINGERPRINT_CONFLICT, None])
-        callback = mock.Mock()
+        provider = _FakeProvider([CONFLICT, APPLY])
+        callback = mock.Mock(side_effect=lambda md: md['items'].append(
+            {'key': 'k', 'value': 'v'}))
 
-        _save(provider, callback)
+        gcp_metadata_save_op(provider, callback)
 
         # Two attempts, each on metadata fetched anew so the retry carries
         # the fingerprint the conflict invalidated.
-        self.assertEqual(provider.gcp_compute.fetches, 2)
         self.assertEqual(callback.call_count, 2)
         self.assertEqual(
-            [body['fingerprint'] for body in provider.gcp_compute.saved_bodies],
-            ['fp-1', 'fp-2'])
+            [b['fingerprint'] for b in provider.gcp_compute.saved_bodies],
+            ['fp-0', 'fp-0'])
+        self.assertEqual(provider.gcp_compute.keys(), {'k': 'v'})
+
+    def test_write_reported_done_but_absent_is_reapplied(self):
+        # The live suite showed operations completing as DONE with the
+        # change missing from the document. A write is only finished once
+        # it can be read back.
+        provider = _FakeProvider([LOST, APPLY])
+
+        modify_or_add_metadata_item(provider, 'k', 'v')
+
+        self.assertEqual(len(provider.gcp_compute.saved_bodies), 2)
+        self.assertEqual(provider.gcp_compute.keys(), {'k': 'v'})
+
+    def test_write_that_never_lands_is_reported_after_retries(self):
+        provider = _FakeProvider([LOST] * 10)
+
+        with self.assertRaises(MetadataWriteNotApplied) as raised:
+            modify_or_add_metadata_item(provider, 'k', 'v')
+
+        self.assertIn('k', str(raised.exception))
+        self.assertEqual(len(provider.gcp_compute.saved_bodies), 10)
 
     def test_other_operation_failures_are_raised_as_typed_errors(self):
         provider = _FakeProvider([OTHER_FAILURE])
-        callback = mock.Mock()
+        callback = mock.Mock(side_effect=lambda md: md['items'].append(
+            {'key': 'k', 'value': 'v'}))
 
         with self.assertRaises(GCPOperationError) as raised:
-            _save(provider, callback)
+            gcp_metadata_save_op(provider, callback)
 
         self.assertEqual(callback.call_count, 1)
         self.assertEqual(raised.exception.codes, ['RESOURCE_NOT_FOUND'])
         self.assertIn("was not found", str(raised.exception))
+
+    def test_a_write_that_changes_nothing_is_not_sent(self):
+        # Every set re-uploads the whole document and moves the fingerprint,
+        # so a no-op (removing an absent key, re-setting the current value)
+        # would only add contention for other writers.
+        provider = _FakeProvider([], items=[{'key': 'k', 'value': 'v'}])
+
+        self.assertFalse(remove_metadata_item(provider, 'absent'))
+        modify_or_add_metadata_item(provider, 'k', 'v')
+
+        self.assertEqual(provider.gcp_compute.saved_bodies, [])
+
+    def test_remove_is_reapplied_until_the_key_is_gone(self):
+        provider = _FakeProvider([LOST, APPLY],
+                                 items=[{'key': 'k', 'value': 'v'}])
+
+        self.assertTrue(remove_metadata_item(provider, 'k'))
+
+        self.assertEqual(len(provider.gcp_compute.saved_bodies), 2)
+        self.assertEqual(provider.gcp_compute.keys(), {})
+
+    def test_add_is_satisfied_by_its_own_value_already_present(self):
+        # A retry must not append a second copy if the earlier attempt did
+        # land after all.
+        provider = _FakeProvider([], items=[{'key': 'k', 'value': 'v'}])
+
+        add_metadata_item(provider, 'k', 'v')
+
+        self.assertEqual(provider.gcp_compute.saved_bodies, [])
+
+    def test_add_rejects_a_key_someone_else_holds(self):
+        provider = _FakeProvider([], items=[{'key': 'k', 'value': 'theirs'}])
+
+        with self.assertRaises(DuplicateResourceException):
+            add_metadata_item(provider, 'k', 'mine')
+
+        self.assertEqual(provider.gcp_compute.saved_bodies, [])