Check if record is stale after bulk pull

We need to check if the results from a bulk_pull are stale because
the resource might have been updated and concurrently pushed from
the server while the bulk_pull query was being fulfilled.

Change-Id: I755a1cb2e0037ec2316161a09ad462bc4b09f397
Closes-Bug: #1707699

neutron/agent/resource_cache.py

@@ -75,6 +75,11 @@ class RemoteResourceCache(object):
         resources = self._puller.bulk_pull(context, rtype,
                                            filter_kwargs=filter_kwargs)
         for resource in resources:
+            if self._is_stale(rtype, resource):
+                # if the server was slow enough to respond the object may have
+                # been updated already and pushed to us in another thread.
+                LOG.debug("Ignoring stale update for %s: %s", rtype, resource)
+                continue
             self._type_cache(rtype)[resource.id] = resource
         LOG.debug("%s resources returned for queries %s", len(resources),
                   query_ids)

neutron/tests/unit/agent/test_resource_cache.py

@@ -71,6 +71,16 @@ class RemoteResourceCacheTestCase(base.BaseTestCase):
         self._pullmock.bulk_pull.assert_called_once_with(
             mock.ANY, 'goose', filter_kwargs={'id': (67, )})
 
+    def test_bulk_pull_doesnt_wipe_out_newer_data(self):
+        self.rcache.record_resource_update(
+            self.ctx, 'goose', OVOLikeThing(1, revision_number=5))
+        updated = OVOLikeThing(1)
+        updated.revision_number = 1  # older revision number
+        self._pullmock.bulk_pull.return_value = [updated]
+        self.rcache._flood_cache_for_query('goose', id=(1,),)
+        self.assertEqual(
+             5, self.rcache.get_resource_by_id('goose', 1).revision_number)
+
     def test_get_resources(self):
         geese = [OVOLikeThing(3, size='large'), OVOLikeThing(5, size='medium'),
                  OVOLikeThing(4, size='large'), OVOLikeThing(6, size='small')]