policy: Fix stream generation accounting

jrajahalme · jrajahalme · commit 18261d2c12bc · 2026-04-18T20:35:05.000+02:00
Stream generation accounting has to be shared between NPDS and NPRDS
streams, so that the handoff works as designed, but no other xDS
protocols (e.g., NPHDS) should interfere with the stream generation
accounting. Solve this by defining the stream generation number as a
static member of NetworkPolicyMapImpl and updating it from the already
established transport connected/closed callbacks.

Adjust tests to work with the new shape where the generation numbers do
not start from 1 for each NetworkPolicyMapImpl instance, but increase
monotonically for each established NPDS/NPRDS stream.

Signed-off-by: Jarno Rajahalme &lt;jarno@isovalent.com&gt;
diff --git a/cilium/grpc_subscription.cc b/cilium/grpc_subscription.cc
@@ -43,13 +43,34 @@ namespace Cilium {
 
 namespace {
 
-constexpr uint64_t FirstStreamGeneration = 1;
-
 class StreamTrackedGrpcMux {
 public:
+  StreamTrackedGrpcMux(std::function<void()> on_transport_established,
+                       std::function<void()> on_transport_close)
+      : on_transport_established_(std::move(on_transport_established)),
+        on_transport_close_(std::move(on_transport_close)) {}
+
   virtual ~StreamTrackedGrpcMux() = default;
-  virtual uint64_t streamGeneration() const = 0;
-  virtual bool streamConnected() const = 0;
+
+  bool streamConnected() const { return stream_connected_; }
+
+  void onStreamEstablished() {
+    stream_connected_ = true;
+    on_transport_established_();
+  }
+
+  void onEstablishmentFailure() {
+    const bool was_connected = stream_connected_;
+    stream_connected_ = false;
+    if (was_connected) {
+      on_transport_close_();
+    }
+  }
+
+private:
+  bool stream_connected_{false};
+  std::function<void()> on_transport_established_;
+  std::function<void()> on_transport_close_;
 };
 
 class SotwGrpcMuxImpl : public Config::GrpcMuxImpl, public StreamTrackedGrpcMux {
@@ -58,37 +79,18 @@ class SotwGrpcMuxImpl : public Config::GrpcMuxImpl, public StreamTrackedGrpcMux
                   std::function<void()> on_transport_established,
                   std::function<void()> on_transport_close)
       : Config::GrpcMuxImpl(grpc_mux_context, skip_subsequent_node),
-        on_transport_established_(std::move(on_transport_established)),
-        on_transport_close_(std::move(on_transport_close)) {}
-
+        StreamTrackedGrpcMux(std::move(on_transport_established), std::move(on_transport_close)) {}
   ~SotwGrpcMuxImpl() override = default;
 
   void onStreamEstablished() override {
-    stream_connected_ = true;
-    ++stream_generation_;
     Config::GrpcMuxImpl::onStreamEstablished();
-    if (on_transport_established_) {
-      on_transport_established_();
-    }
+    StreamTrackedGrpcMux::onStreamEstablished();
   }
 
   void onEstablishmentFailure(bool next_attempt_may_send_initial_resource_version) override {
-    const bool was_connected = stream_connected_;
-    stream_connected_ = false;
     Config::GrpcMuxImpl::onEstablishmentFailure(next_attempt_may_send_initial_resource_version);
-    if (was_connected && on_transport_close_) {
-      on_transport_close_();
-    }
+    StreamTrackedGrpcMux::onEstablishmentFailure();
   }
-
-  uint64_t streamGeneration() const override { return stream_generation_; }
-  bool streamConnected() const override { return stream_connected_; }
-
-private:
-  uint64_t stream_generation_{0};
-  bool stream_connected_{false};
-  std::function<void()> on_transport_established_;
-  std::function<void()> on_transport_close_;
 };
 
 class DeltaGrpcMuxImpl : public Config::NewGrpcMuxImpl, public StreamTrackedGrpcMux {
@@ -97,37 +99,19 @@ class DeltaGrpcMuxImpl : public Config::NewGrpcMuxImpl, public StreamTrackedGrpc
                             std::function<void()> on_transport_established,
                             std::function<void()> on_transport_close)
       : Config::NewGrpcMuxImpl(grpc_mux_context),
-        on_transport_established_(std::move(on_transport_established)),
-        on_transport_close_(std::move(on_transport_close)) {}
+        StreamTrackedGrpcMux(std::move(on_transport_established), std::move(on_transport_close)) {}
 
   ~DeltaGrpcMuxImpl() override = default;
 
   void onStreamEstablished() override {
-    stream_connected_ = true;
-    ++stream_generation_;
     Config::NewGrpcMuxImpl::onStreamEstablished();
-    if (on_transport_established_) {
-      on_transport_established_();
-    }
+    StreamTrackedGrpcMux::onStreamEstablished();
   }
 
   void onEstablishmentFailure(bool next_attempt_may_send_initial_resource_version) override {
-    const bool was_connected = stream_connected_;
-    stream_connected_ = false;
     Config::NewGrpcMuxImpl::onEstablishmentFailure(next_attempt_may_send_initial_resource_version);
-    if (was_connected && on_transport_close_) {
-      on_transport_close_();
-    }
+    StreamTrackedGrpcMux::onEstablishmentFailure();
   }
-
-  uint64_t streamGeneration() const override { return stream_generation_; }
-  bool streamConnected() const override { return stream_connected_; }
-
-private:
-  uint64_t stream_generation_{0};
-  bool stream_connected_{false};
-  std::function<void()> on_transport_established_;
-  std::function<void()> on_transport_close_;
 };
 
 // service RPC method fully qualified names.
@@ -233,20 +217,6 @@ envoy::config::core::v3::ConfigSource getCiliumXDSAPIConfig(bool use_delta_xds =
 
 envoy::config::core::v3::ConfigSource cilium_xds_api_config = getCiliumXDSAPIConfig();
 
-uint64_t grpcStreamGeneration(Config::Subscription* subscription) {
-  auto* sub = dynamic_cast<Config::GrpcSubscriptionImpl*>(subscription);
-  if (!sub) {
-    return FirstStreamGeneration;
-  }
-
-  auto* grpc_mux = dynamic_cast<StreamTrackedGrpcMux*>(sub->grpcMux().get());
-  if (grpc_mux == nullptr) {
-    return FirstStreamGeneration;
-  }
-
-  return grpc_mux->streamGeneration();
-}
-
 bool grpcStreamConnected(Config::Subscription* subscription) {
   auto* sub = dynamic_cast<Config::GrpcSubscriptionImpl*>(subscription);
   if (!sub) {
diff --git a/cilium/grpc_subscription.h b/cilium/grpc_subscription.h
@@ -25,12 +25,6 @@ subscribe(const std::string& type_url, Server::Configuration::CommonFactoryConte
           std::function<void()> on_transport_established = {},
           std::function<void()> on_transport_close = {});
 
-// Returns a monotonic stream generation for Cilium subscriptions.
-// Value 0 is reserved for policy-map detection of the initial stream and may be returned for
-// tracked gRPC subscriptions before any stream has been established.
-// Non-gRPC subscriptions and subscriptions without stream tracking are treated as generation 1.
-uint64_t grpcStreamGeneration(Config::Subscription* subscription);
-
 // Returns whether a tracked gRPC subscription currently has an established transport.
 bool grpcStreamConnected(Config::Subscription* subscription);
 
diff --git a/cilium/network_policy.cc b/cilium/network_policy.cc
@@ -411,12 +411,8 @@ class NetworkPolicyMapImpl : public Envoy::Config::SubscriptionCallbacks,
   }
 
 protected:
-  uint64_t streamGeneration() const {
-    return stream_generation_override_for_test_ != 0 ? stream_generation_override_for_test_
-                                                     : grpcStreamGeneration(subscription_.get());
-  }
-
-  void resetStreamForTest() { stream_generation_override_for_test_ = streamGeneration() + 1; }
+  uint64_t streamGeneration() const { return subscription_stream_generation_; }
+  void resetStreamForTest() { subscription_stream_generation_++; }
 
   // run the given function after all the threads have scheduled
   void runAfterAllThreads(std::function<void()> cb) const {
@@ -454,6 +450,8 @@ class NetworkPolicyMapImpl : public Envoy::Config::SubscriptionCallbacks,
   }
 
   void onSubscriptionTransportEstablished(uint64_t subscription_id) {
+    ++subscription_stream_generation_;
+
     if (subscription_id != subscription_id_) {
       return;
     }
@@ -561,10 +559,8 @@ class NetworkPolicyMapImpl : public Envoy::Config::SubscriptionCallbacks,
       transport_factory_context_;
 
   std::unique_ptr<Envoy::Config::Subscription> subscription_;
+  static uint64_t subscription_stream_generation_;
   NetworkPolicyMap::SubscriptionFactoryForTest subscription_factory_for_test_;
-  // Test-only override used to simulate a restarted NPDS stream when the test subscription does
-  // not expose a new underlying gRPC stream generation.
-  uint64_t stream_generation_override_for_test_{0};
 
   ProtobufTypes::MessagePtr dumpNetworkPolicyConfigs(const Matchers::StringMatcher& name_matcher);
   Server::ConfigTracker::EntryOwnerPtr config_tracker_entry_;
@@ -576,6 +572,7 @@ class NetworkPolicyMapImpl : public Envoy::Config::SubscriptionCallbacks,
 };
 
 uint64_t NetworkPolicyMapImpl::instance_id_ = 0;
+uint64_t NetworkPolicyMapImpl::subscription_stream_generation_ = 1;
 
 IpAddressPair::IpAddressPair(const cilium::NetworkPolicy& proto) {
   for (const auto& ip_addr : proto.endpoint_ips()) {
diff --git a/tests/cilium_network_policy_test.cc b/tests/cilium_network_policy_test.cc
@@ -1801,8 +1801,10 @@ TEST_F(CiliumNetworkPolicyDeltaTest, SameStreamSelectorOnlyUpdateUsesLatestSelec
 
   const auto old_policy = policyInstanceShared("10.1.2.3");
   ASSERT_NE(nullptr, old_policy);
+  const auto initial_stream_generation = selectorStreamGenerationForTest(*old_policy);
 
-  EXPECT_EQ(1, selectorStreamGenerationForTest(*old_policy));
+  EXPECT_GT(initial_stream_generation, 0);
+  EXPECT_EQ(initial_stream_generation, selectorStreamGenerationForTest(*old_policy));
   EXPECT_EQ(1, selectorVersionForTest(*old_policy));
 
   EXPECT_NO_THROW(deltaUpdateFromYaml(R"EOF(system_version_info: "2"
@@ -1815,7 +1817,7 @@ TEST_F(CiliumNetworkPolicyDeltaTest, SameStreamSelectorOnlyUpdateUsesLatestSelec
       remote_identities: [ 44 ]
 )EOF"));
 
-  EXPECT_EQ(1, selectorStreamGenerationForTest(*old_policy));
+  EXPECT_EQ(initial_stream_generation, selectorStreamGenerationForTest(*old_policy));
   EXPECT_EQ(2, selectorVersionForTest(*old_policy));
   EXPECT_TRUE(ingressAllowed("10.1.2.3", 44, 80));
   EXPECT_FALSE(ingressAllowed("10.1.2.3", 43, 80));
@@ -1852,8 +1854,10 @@ TEST_F(CiliumNetworkPolicyDeltaTest, NewStreamKeepsOldPolicyPinnedToOldSelectorS
 
   const auto old_policy = policyInstanceShared("10.1.2.3");
   ASSERT_NE(nullptr, old_policy);
+  const auto initial_stream_generation = selectorStreamGenerationForTest(*old_policy);
 
-  EXPECT_EQ(1, selectorStreamGenerationForTest(*old_policy));
+  EXPECT_GT(initial_stream_generation, 0);
+  EXPECT_EQ(initial_stream_generation, selectorStreamGenerationForTest(*old_policy));
   EXPECT_EQ(1, selectorVersionForTest(*old_policy));
 
   EXPECT_NO_THROW(deltaUpdateFromYaml(R"EOF(system_version_info: "1"
@@ -1866,7 +1870,7 @@ TEST_F(CiliumNetworkPolicyDeltaTest, NewStreamKeepsOldPolicyPinnedToOldSelectorS
       remote_identities: [ 45 ]
 )EOF"));
 
-  EXPECT_EQ(1, selectorStreamGenerationForTest(*old_policy));
+  EXPECT_EQ(initial_stream_generation, selectorStreamGenerationForTest(*old_policy));
   EXPECT_EQ(2, selectorVersionForTest(*old_policy));
 
   resetStreamForTest();
@@ -1903,9 +1907,9 @@ TEST_F(CiliumNetworkPolicyDeltaTest, NewStreamKeepsOldPolicyPinnedToOldSelectorS
   ASSERT_NE(nullptr, new_policy);
   EXPECT_NE(old_policy.get(), new_policy.get());
 
-  EXPECT_EQ(1, selectorStreamGenerationForTest(*old_policy));
+  EXPECT_EQ(initial_stream_generation, selectorStreamGenerationForTest(*old_policy));
   EXPECT_EQ(2, selectorVersionForTest(*old_policy));
-  EXPECT_EQ(2, selectorStreamGenerationForTest(*new_policy));
+  EXPECT_EQ(initial_stream_generation + 1, selectorStreamGenerationForTest(*new_policy));
   EXPECT_EQ(3, selectorVersionForTest(*new_policy));
   EXPECT_TRUE(ingressAllowed("10.1.2.3", 44, 80));
   EXPECT_FALSE(ingressAllowed("10.1.2.3", 43, 80));
