Fix indefinite port_settings_* failure cases

dpd-client/tests/chaos_tests/port_settings.rs

@@ -13,7 +13,7 @@ use crate::chaos_tests::harness;
 use crate::chaos_tests::util::HttpResponseCheck;
 use crate::chaos_tests::util::IpRng;
 
-use anyhow::bail;
+use anyhow::{Context, bail};
 use asic::chaos::{AsicConfig, Chaos, TableChaos};
 use asic::table_chaos;
 use common::table::TableType;
@@ -49,6 +49,9 @@ const LINK_CREATE: LinkCreate = LinkCreate {
     allow_ddm_traffic: false,
 };
 
+const TAG1: &str = "chaos1";
+const TAG2: &str = "chaos2";
+
 #[cfg(test)]
 mod retry {
     use std::future::Future;
@@ -519,8 +522,92 @@ fn random_port_settings() -> PortSettings {
     }
 }
 
-const TAG1: &str = "chaos1";
-const TAG2: &str = "chaos2";
+/// A simplified version of txn_sweep that ensures `port_settings_*`
+/// functions can succeed after partial failures.
+#[tokio::test]
+async fn settings_eventually_reconcile() -> anyhow::Result<()> {
+    let mut apply = 0;
+    let mut clear = 0;
+    let mut get = 0;
+
+    // Limit the test duration so a failure doesn't run indefinitely in CI.
+    //
+    // This is mathematically capable of flaking due to RNG, but the
+    // timeout should be high enough that it's safe for CI.
+    let status = tokio::time::timeout(Duration::from_secs(120), async {
+        self::settings_eventually_reconcile_unbounded(
+            &mut apply, &mut clear, &mut get,
+        )
+        .await
+    })
+    .await
+    .context("Timed out waiting for successful reconciliation");
+
+    println!(
+        "
+Reconciliation retries:
+    - port_settings_apply: {apply}
+    - port_settings_clear: {clear}
+    - port_settings_get: {get}
+"
+    );
+
+    status??;
+    Ok(())
+}
+
+async fn settings_eventually_reconcile_unbounded(
+    apply_ct: &mut usize,
+    clear_ct: &mut usize,
+    get_ct: &mut usize,
+) -> anyhow::Result<()> {
+    let config = AsicConfig::uniform_set(TESTING_RADIX, 0.4);
+    let (_guard, client) =
+        harness::init_harness("settings_eventually_reconcile", &config);
+    let mut rng = IpRng::new(1046);
+
+    let port_id: PortId = "qsfp0".parse()?;
+    let link_id = LinkId(0);
+
+    let settings = PortSettings {
+        links: [(
+            link_id.to_string(),
+            LinkSettings {
+                params: LINK_CREATE,
+                addrs: vec![rng.unique_ipv4().into(), rng.unique_ipv6().into()],
+            },
+        )]
+        .into_iter()
+        .collect(),
+    };
+
+    for _ in 0..3 {
+        while client
+            .port_settings_apply(&port_id, Some(TAG1), &settings)
+            .await
+            .is_err()
+        {
+            *apply_ct += 1;
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+
+        while client.port_settings_clear(&port_id, Some(TAG1)).await.is_err() {
+            *clear_ct += 1;
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+
+        while !client
+            .port_settings_get(&port_id, Some(TAG1))
+            .await
+            .is_ok_and(|s| s.links.is_empty())
+        {
+            *get_ct += 1;
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    }
+
+    Ok(())
+}
 
 /// Verifies tagged port_settings_apply actions don't affect
 /// resources from other tags.
@@ -883,6 +970,72 @@ async fn apply_fails_on_tag_conflict() -> anyhow::Result<()> {
     Ok(())
 }
 
+/// A pathological sequence of table errors in dpd must
+/// not poison future valid operations.
+///
+/// This test port_settings_applies two addresses. The IPv6
+/// table op fails, and then the IPv4 table op in the rollback
+/// fails. After that rollback failure, the problematic IPv4
+/// table slot should still be usable.
+#[tokio::test]
+async fn partial_failures_are_recoverable() -> anyhow::Result<()> {
+    let conf = AsicConfig {
+        radix: TESTING_RADIX,
+        table_entry_add: table_chaos!((TableType::PortAddrIpv6, 1.0)),
+        table_entry_del: table_chaos!((TableType::PortAddrIpv4, 1.0)),
+        ..Default::default()
+    };
+
+    let (_guard, client) =
+        harness::init_harness("partial_failures_are_recoverable", &conf);
+    let mut rng = IpRng::new(731);
+    let port_id: PortId = "qsfp0".parse()?;
+    let link_id = LinkId(0);
+
+    let addrs = TestAddrs::new(
+        &mut rng,
+        TAG1.to_string(),
+        &client,
+        port_id.clone(),
+        link_id,
+    );
+
+    let apply_err = addrs
+        .apply_addrs()
+        .await
+        .expect_err("Apply should fail because IPv6 table ops fail");
+    assert!(
+        self::is_rollback_error(&apply_err),
+        "IPv4 addr couldn't be rolled back from table"
+    );
+
+    // Not sure this is the best behavior between apply and clear amid rollback
+    // failures, but this test at least ensures we can reclaim the entry later.
+    let empty = client.port_settings_clear(&port_id, Some(TAG1)).await?;
+    assert!(
+        empty.links.is_empty(),
+        "Unsuccessful apply was rolled back. The IPv4 entry is still stuck in the table because rollback partially failed, but clear doesn't know about that."
+    );
+
+    let mut v4_only = addrs.settings();
+    for link in v4_only.links.values_mut() {
+        link.addrs.retain(|a| a.is_ipv4());
+    }
+
+    let settings = client
+        .port_settings_apply(&port_id, Some(TAG2), &v4_only)
+        .await
+        .context("Apply should succeed because we can at least overwrite the IPv4 table entry")?;
+
+    let registered_v4 = settings
+        .links
+        .values()
+        .any(|link| link.addrs.contains(&addrs.v4_entry.addr.into()));
+    assert!(registered_v4, "Address was reclaimed");
+
+    Ok(())
+}
+
 /// This struct simplifies repetitive CRUD operations
 /// on tagged links with random address registrations.
 struct TestAddrs<'a> {
@@ -923,6 +1076,22 @@ impl<'a> TestAddrs<'a> {
         Ok(())
     }
 
+    /// Creates a [`PortSettings`] instance for these addresses.
+    fn settings(&self) -> PortSettings {
+        PortSettings {
+            links: HashMap::from([(
+                self.link_id.to_string(),
+                LinkSettings {
+                    params: LINK_CREATE,
+                    addrs: vec![
+                        self.v4_entry.addr.into(),
+                        self.v6_entry.addr.into(),
+                    ],
+                },
+            )]),
+        }
+    }
+
     /// Adds both tagged addresses to this link using dpd's `port_settings_apply` endpoint.
     async fn apply_addrs(
         &self,
@@ -931,18 +1100,7 @@ impl<'a> TestAddrs<'a> {
             .port_settings_apply(
                 &self.port_id,
                 Some(&self.v4_entry.tag),
-                &PortSettings {
-                    links: HashMap::from([(
-                        self.link_id.to_string(),
-                        LinkSettings {
-                            params: LINK_CREATE,
-                            addrs: vec![
-                                self.v4_entry.addr.into(),
-                                self.v6_entry.addr.into(),
-                            ],
-                        },
-                    )]),
-                },
+                &self.settings(),
             )
             .await?;
 

dpd/src/api_server.rs

@@ -1710,7 +1710,7 @@ impl DpdApi for DpdApiImpl {
         route::reset_ipv4_tag(switch, &tag).await;
         route::reset_ipv6_tag(switch, &tag).await;
         switch
-            .clear_link_addresses(Some(&tag))
+            .clear_link_state(Some(&tag))
             .map(|_| HttpResponseUpdatedNoContent())
             .map_err(|e| e.into())
     }
@@ -1734,7 +1734,7 @@ impl DpdApi for DpdApiImpl {
             error!(switch.log, "failed to reset route data: {:?}", e);
             err = Some(e);
         }
-        if let Err(e) = switch.clear_link_state() {
+        if let Err(e) = switch.clear_link_state(None) {
             error!(switch.log, "failed to clear all link state: {:?}", e);
             err = Some(e);
         }