[PATCH proxmox-ve-rs v2 01/27] ve-config: sdn: add microseg signature-identity engine

[Hannes Laimer <h.laimer@proxmox.com>]

Each interface carries a set of groups, but the wire has room for a
single 16-bit id. The engine partitions the realized group sets into
equivalence classes that no rule can tell apart and numbers those
classes, so the id space scales with policy-distinguishable sets rather
than every set combination, and all hosts agree on what a number means
because the assignment is rendered once and replicated.

Ids must stay stable while the config changes around them: a number on
the wire has no transactional boundary (it is stamped on one host and
looked up on another), so rebinding it to an unrelated class could give
an in-flight packet the wrong verdict. Allocation is therefore
fresh-first from a counter, splits mint new numbers, merges retire the
losing number, and a retired number only becomes reusable after a
quarantine long enough that no packet stamped under its old meaning can
still exist. Reuse starts only once the fresh space is exhausted, so
exhaustion requires more than 65534 classes live or freshly retired at
the same time, not merely a long-lived, frequently edited cluster.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 .../src/sdn/microseg/identity.rs              | 624 ++++++++++++++++++
 proxmox-ve-config/src/sdn/microseg/mod.rs     |   9 +
 proxmox-ve-config/src/sdn/mod.rs              |   1 +
 3 files changed, 634 insertions(+)
 create mode 100644 proxmox-ve-config/src/sdn/microseg/identity.rs
 create mode 100644 proxmox-ve-config/src/sdn/microseg/mod.rs

diff --git a/proxmox-ve-config/src/sdn/microseg/identity.rs b/proxmox-ve-config/src/sdn/microseg/identity.rs
new file mode 100644
index 0000000..ab700ae
--- /dev/null
+++ b/proxmox-ve-config/src/sdn/microseg/identity.rs
@@ -0,0 +1,624 @@
+//! Signature-based identity allocation for microseg.
+//!
+//! An *identity* is not one group-set but a class of group-sets that no policy can tell apart. Two
+//! sets share an id when every policy's src and dst predicate fires the same way on both. The class
+//! carries an *id* (stamped on the wire) and a *representative*, the smallest member set in sort
+//! order, so a later split or merge stays deterministic and ids stay stable.
+//!
+//! Allocation is **fresh-first**. Ids come from a counter that only increases, so a number is never
+//! reused while unissued space remains. A number on the wire has no transactional boundary, so
+//! rebinding it to an unrelated class could give an in-flight packet the wrong verdict. Splits
+//! issue fresh ids. Merges retire the fused-away id into a time-based quarantine
+//! ([`RETIRE_QUARANTINE_SECS`]). Exhaustion needs more than 65534 classes live or freshly retired
+//! at once.
+//!
+//! The registry is the only stored state, holding the id assignment that cannot be recomputed. The
+//! partition is always re-derived from the current marks and policies.
+
+use std::collections::{BTreeMap, BTreeSet, HashMap};
+
+use serde::{Deserialize, Serialize};
+
+/// The canonical group-set assigned to an interface, as sorted, deduped base-group marks.
+/// `BTreeSet`'s `Ord` compares the sorted elements in order, which is exactly the representative
+/// order.
+pub type GroupSet = BTreeSet<u16>;
+
+/// Reserved id for unstamped traffic. A frame that arrives without a tag (a gateway, DHCP/ARP, a
+/// cross-node frame with no GBP tag) reads as this id. Never issued to a class, so a real group set
+/// (even one no policy matches) never collides with it and never inherits its `untagged` allows.
+/// Its representative in the rule fold is the built-in `untagged` group.
+pub const UNTAGGED_ID: u16 = 0;
+
+/// How long a retired id stays quarantined before it may be re-issued, in seconds.
+///
+/// Reusing a number is only safe once no packet stamped under its old meaning can still exist. The
+/// generous margin also covers nodes that apply a render late. A node cut off from pmxcfs for
+/// longer than this while still passing traffic can stamp stale ids into a reused number. That is
+/// accepted, a cluster in that state is broken in worse ways.
+pub const RETIRE_QUARANTINE_SECS: u64 = 60 * 60;
+
+/// How a predicate decides whether it fires on an interface's group-set.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PredicateKind {
+    /// Fires when the set shares at least one group with the predicate (intersection non-empty).
+    Any,
+    /// Fires when the set contains every group of the predicate (predicate is a subset of the set).
+    All,
+    /// Fires only when the set equals the predicate exactly.
+    Exact,
+}
+
+/// A predicate over base groups, the src or dst half of a policy. An empty `groups` is degenerate
+/// (`Any` fires on nothing, `All` fires on everything) and is rejected by config validation, not
+/// here.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Predicate {
+    pub kind: PredicateKind,
+    pub groups: GroupSet,
+}
+
+impl Predicate {
+    pub fn new(kind: PredicateKind, groups: impl IntoIterator<Item = u16>) -> Self {
+        Self {
+            kind,
+            groups: groups.into_iter().collect(),
+        }
+    }
+
+    /// Whether this predicate fires on `set`.
+    pub fn fires(&self, set: &GroupSet) -> bool {
+        match self.kind {
+            PredicateKind::Any => self.groups.intersection(set).next().is_some(),
+            PredicateKind::All => self.groups.is_subset(set),
+            PredicateKind::Exact => &self.groups == set,
+        }
+    }
+}
+
+/// A policy adds two bits to every set's signature, whether the set is a valid *source* (`src`
+/// fires) and whether it is a valid *destination* (`dst` fires). The allow/deny verdict is
+/// downstream of identity and lives in the fold, not here (matchability is not the verdict).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Policy {
+    pub src: Predicate,
+    pub dst: Predicate,
+}
+
+/// How a set fires against each policy in order, as `[p0.src, p0.dst, p1.src, ...]`. Only equality
+/// matters, equal signatures are interchangeable in the rule map. The policy order must be fixed
+/// for signatures within one partition to be comparable, so callers pass a canonical order.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct Signature(Vec<bool>);
+
+impl Signature {
+    /// Whether no policy's src or dst predicate fires. Such sets look the same to every policy and
+    /// share a single class, but that class still gets its own id, distinct from the [`UNTAGGED_ID`]
+    /// of unstamped traffic.
+    pub fn is_all_false(&self) -> bool {
+        self.0.iter().all(|fired| !fired)
+    }
+}
+
+/// Compute a set's signature against `policies` (canonical order).
+pub fn signature(set: &GroupSet, policies: &[Policy]) -> Signature {
+    let mut bits = Vec::with_capacity(policies.len() * 2);
+    for policy in policies {
+        bits.push(policy.src.fires(set));
+        bits.push(policy.dst.fires(set));
+    }
+    Signature(bits)
+}
+
+/// Each live id pinned to its representative set, plus the id counter. Only the non-recomputable
+/// part, the id assignment, is kept. The partition is derived from the policies on demand.
+/// Serialized into the running config so every host reads the same id-to-class naming.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Registry {
+    next: u16,
+    /// id to representative set (lex-min member of the class). Serialized with string keys (see
+    /// [`u16_key_serde`]) so the Perl/JSON round trip the running config takes does not choke on
+    /// integer map keys.
+    #[serde(with = "u16_key_serde")]
+    classes: BTreeMap<u16, GroupSet>,
+    /// Recently retired id to unix time of retirement. A number in here is inside the reclamation
+    /// quarantine, no longer live, but packets stamped under its old meaning may still exist. So it
+    /// must not be re-issued yet. Entries older than [`RETIRE_QUARANTINE_SECS`] are dropped by the
+    /// next [`allocate`], turning the number into plain free space.
+    #[serde(
+        default,
+        with = "u16_key_serde",
+        skip_serializing_if = "BTreeMap::is_empty"
+    )]
+    retired: BTreeMap<u16, u64>,
+}
+
+/// Serialize a `u16`-keyed map with string keys. JSON forces object keys to strings, and perlmod
+/// (the running config's Perl round trip) does not parse a string key back into a `u16`, so we
+/// stringify on the way out and parse on the way in. The values are plain numbers and survive the
+/// round trip as-is. serde_json reads either form, so this stays compatible with an existing
+/// running config.
+mod u16_key_serde {
+    use std::collections::BTreeMap;
+
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    pub fn serialize<S: Serializer, V: Serialize>(
+        map: &BTreeMap<u16, V>,
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        let stringified: BTreeMap<String, &V> = map
+            .iter()
+            .map(|(id, value)| (id.to_string(), value))
+            .collect();
+        stringified.serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>, V: Deserialize<'de>>(
+        deserializer: D,
+    ) -> Result<BTreeMap<u16, V>, D::Error> {
+        BTreeMap::<String, V>::deserialize(deserializer)?
+            .into_iter()
+            .map(|(id, value)| {
+                id.parse::<u16>()
+                    .map(|id| (id, value))
+                    .map_err(serde::de::Error::custom)
+            })
+            .collect()
+    }
+}
+
+impl Default for Registry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Registry {
+    /// An empty registry. The first issued id is 1 (0 is the untagged id).
+    pub fn new() -> Self {
+        Self {
+            next: 1,
+            classes: BTreeMap::new(),
+            retired: BTreeMap::new(),
+        }
+    }
+
+    /// Reconstruct a registry from its persisted parts (e.g. the running-config).
+    pub fn from_parts(
+        next: u16,
+        classes: BTreeMap<u16, GroupSet>,
+        retired: BTreeMap<u16, u64>,
+    ) -> Self {
+        Self {
+            next,
+            classes,
+            retired,
+        }
+    }
+
+    pub fn next(&self) -> u16 {
+        self.next
+    }
+
+    /// id to representative set for every live (non-retired) class.
+    pub fn classes(&self) -> &BTreeMap<u16, GroupSet> {
+        &self.classes
+    }
+
+    /// Recently retired id to retirement time, still inside the reclamation quarantine.
+    pub fn retired(&self) -> &BTreeMap<u16, u64> {
+        &self.retired
+    }
+
+    /// The id of the class a set belongs to, by signature match. Every realized set has a class,
+    /// including one no policy matches (it shares the all-false class), so this resolves them all.
+    /// Returns `None` only for a set no current class carries, a not-yet-rendered set on a config
+    /// whose render has not landed, so the caller fails closed.
+    pub fn id_of(&self, set: &GroupSet, policies: &[Policy]) -> Option<u16> {
+        let sig = signature(set, policies);
+        self.classes
+            .iter()
+            .find(|(_, rep)| signature(rep, policies) == sig)
+            .map(|(&id, _)| id)
+    }
+}
+
+/// Recompute the registry from the realized sets and policies, fresh-first against `prev`.
+///
+/// Every realized set is partitioned by signature, including the sets no policy matches. They share
+/// the all-false signature and so collapse into a single class with its own id, kept distinct from
+/// the [`UNTAGGED_ID`] of unstamped traffic so an unmatched set never inherits `untagged` allows.
+/// Each class keeps the id of the previous class whose representative still lands in it
+/// (split-stable), and a merge keeps the lex-min representative's id and retires the rest.
+/// Brand-new classes issue from the counter while fresh space remains. Once it is exhausted they
+/// reuse the lowest number that is neither live nor still inside the retirement quarantine (see
+/// [`RETIRE_QUARANTINE_SECS`]). `now` is the render time (unix seconds), the clock the quarantine
+/// runs on. This must be fed the *previous* registry and never a fresh one (that would renumber the
+/// whole cluster).
+pub fn allocate(
+    prev: &Registry,
+    sets: &BTreeSet<GroupSet>,
+    policies: &[Policy],
+    now: u64,
+) -> anyhow::Result<Registry> {
+    let mut by_sig: HashMap<Signature, BTreeSet<GroupSet>> = HashMap::new();
+    for set in sets {
+        by_sig
+            .entry(signature(set, policies))
+            .or_default()
+            .insert(set.clone());
+    }
+
+    // Which previous ids still land in each live class, re-evaluating their representative under
+    // the *current* policies. A prev class whose signature no longer has any realized set is simply
+    // not added, it retires into the quarantine below.
+    let mut claims: HashMap<Signature, Vec<(u16, GroupSet)>> = HashMap::new();
+    for (&id, rep) in &prev.classes {
+        let sig = signature(rep, policies);
+        if by_sig.contains_key(&sig) {
+            claims.entry(sig).or_default().push((id, rep.clone()));
+        }
+    }
+
+    // assign ids in a deterministic order (by representative) so issuing is reproducible
+    let mut order: Vec<(&Signature, &BTreeSet<GroupSet>)> = by_sig.iter().collect();
+    order.sort_by(|a, b| representative(a.1).cmp(representative(b.1)));
+
+    // Every previously live id, plus the previously retired ids whose quarantine has not passed
+    // yet, is reserved. Entries whose quarantine ended are dropped here, turning their numbers into
+    // plain free space
+    let mut retired: BTreeMap<u16, u64> = prev
+        .retired
+        .iter()
+        .filter(|&(_, &at)| now.saturating_sub(at) < RETIRE_QUARANTINE_SECS)
+        .map(|(&id, &at)| (id, at))
+        .collect();
+    let mut reserved: BTreeSet<u16> = prev.classes.keys().copied().collect();
+    reserved.extend(retired.keys().copied());
+
+    let mut next = prev.next;
+    let mut classes = BTreeMap::new();
+    for (sig, members) in order {
+        let rep = representative(members).clone();
+        let id = match claims.get(sig) {
+            // existing class (possibly a merge), keep the claimant with the lex-min representative
+            // and retire the others by not carrying them
+            Some(claimants) => claimants
+                .iter()
+                .min_by(|a, b| a.1.cmp(&b.1))
+                .map(|(id, _)| *id)
+                .expect("a claims entry is non-empty"),
+            // new class, issue fresh from the counter, falling back to reclamation once the fresh
+            // space is exhausted.
+            None => {
+                if next == UNTAGGED_ID {
+                    reclaim(&reserved, &classes)?
+                } else {
+                    let id = next;
+                    next = next.wrapping_add(1); // u16::MAX wraps to 0, fresh space exhausted
+                    id
+                }
+            }
+        };
+        classes.insert(id, rep);
+    }
+
+    // everything previously live that was not carried retires now and starts its quarantine
+    for &id in prev.classes.keys() {
+        if !classes.contains_key(&id) {
+            retired.insert(id, now);
+        }
+    }
+
+    Ok(Registry {
+        next,
+        classes,
+        retired,
+    })
+}
+
+/// The lowest reusable number, not the untagged id, not reserved (previously live or still
+/// quarantined), not already assigned in this pass. Only called once the fresh space is exhausted,
+/// so anything else is a retired number whose quarantine has ended.
+fn reclaim(reserved: &BTreeSet<u16>, taken: &BTreeMap<u16, GroupSet>) -> anyhow::Result<u16> {
+    (1..=u16::MAX)
+        .find(|id| !reserved.contains(id) && !taken.contains_key(id))
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "microseg identity id space exhausted \
+                 (all 65535 ids are live or recently retired)"
+            )
+        })
+}
+
+/// The lex-min member of a class. `BTreeSet<GroupSet>` iterates its sets in `Ord` (lexicographic)
+/// order, so the first is the representative.
+fn representative(members: &BTreeSet<GroupSet>) -> &GroupSet {
+    members
+        .iter()
+        .next()
+        .expect("a signature class has at least one member")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::PredicateKind::*;
+    use super::*;
+
+    fn set(xs: &[u16]) -> GroupSet {
+        xs.iter().copied().collect()
+    }
+
+    fn pred(kind: PredicateKind, xs: &[u16]) -> Predicate {
+        Predicate::new(kind, xs.iter().copied())
+    }
+
+    fn policy(src: Predicate, dst: Predicate) -> Policy {
+        Policy { src, dst }
+    }
+
+    fn sets(xs: &[&[u16]]) -> BTreeSet<GroupSet> {
+        xs.iter().map(|s| set(s)).collect()
+    }
+
+    #[test]
+    fn fires_semantics() {
+        let s = set(&[2, 3, 4]);
+
+        assert!(pred(Any, &[2]).fires(&s));
+        assert!(pred(Any, &[4, 9]).fires(&s));
+        assert!(!pred(Any, &[9]).fires(&s));
+        assert!(!pred(Any, &[]).fires(&s));
+
+        assert!(pred(All, &[2, 3]).fires(&s));
+        assert!(pred(All, &[2, 3, 4]).fires(&s));
+        assert!(!pred(All, &[2, 9]).fires(&s));
+        assert!(pred(All, &[]).fires(&s));
+
+        assert!(pred(Exact, &[2, 3, 4]).fires(&s));
+        assert!(!pred(Exact, &[2, 3]).fires(&s));
+        assert!(!pred(Exact, &[0]).fires(&set(&[])));
+        assert!(pred(Exact, &[0]).fires(&set(&[0])));
+    }
+
+    #[test]
+    fn equal_signatures_collapse_distinct_signatures_split() {
+        // Only ANY{2} distinguishes, so {2,3} and {2,4} look the same and {5} is separate.
+        let policies = [policy(pred(Any, &[2]), pred(Any, &[9]))];
+        assert_eq!(
+            signature(&set(&[2, 3]), &policies),
+            signature(&set(&[2, 4]), &policies),
+            "no predicate tells (2,3) and (2,4) apart"
+        );
+        assert_ne!(
+            signature(&set(&[2, 3]), &policies),
+            signature(&set(&[5]), &policies),
+        );
+        assert!(signature(&set(&[5]), &policies).is_all_false());
+    }
+
+    #[test]
+    fn fresh_allocation_issues_in_representative_order() {
+        // Two distinct classes. {2,3} sorts before {5} so it issues id 1.
+        let policies = [
+            policy(pred(Any, &[2]), pred(Any, &[9])),
+            policy(pred(Any, &[5]), pred(Any, &[9])),
+        ];
+        let reg = allocate(&Registry::new(), &sets(&[&[2, 3], &[5]]), &policies, 0).unwrap();
+
+        assert_eq!(reg.id_of(&set(&[2, 3]), &policies), Some(1));
+        assert_eq!(reg.id_of(&set(&[5]), &policies), Some(2));
+        assert_eq!(reg.next(), 3);
+        // {7} matches no predicate, here no all-false set was realized, so there is no class for it
+        // and it resolves to nothing
+        assert_eq!(reg.id_of(&set(&[7]), &policies), None);
+        // {9} has a real signature (it matches a dst predicate) but was never realized as a class,
+        // so it resolves to nothing
+        assert_eq!(reg.id_of(&set(&[9]), &policies), None);
+    }
+
+    #[test]
+    fn unmatched_set_gets_its_own_id_not_the_untagged_id() {
+        // A rule references group 2 only. A NIC assigned the unreferenced group 5 has an all-false
+        // signature. It must resolve to its own real id, not the untagged id, so it does not
+        // inherit the allows that an untagged source carries.
+        let policies = [policy(pred(Any, &[0]), pred(Any, &[2]))]; // untagged -> web
+        let reg = allocate(&Registry::new(), &sets(&[&[5]]), &policies, 0).unwrap();
+
+        let id = reg
+            .id_of(&set(&[5]), &policies)
+            .expect("an unmatched realized set still has a class");
+        assert_ne!(
+            id, UNTAGGED_ID,
+            "an unmatched set must not alias onto the untagged id"
+        );
+
+        // two distinct unmatched sets share the one all-false class
+        let reg = allocate(&Registry::new(), &sets(&[&[5], &[6]]), &policies, 0).unwrap();
+        assert_eq!(
+            reg.classes().len(),
+            1,
+            "all unmatched sets collapse to one class"
+        );
+        assert_eq!(
+            reg.id_of(&set(&[5]), &policies),
+            reg.id_of(&set(&[6]), &policies)
+        );
+
+        // the untagged source still fires the untagged rule, the unmatched set does not
+        assert!(pred(Any, &[0]).fires(&set(&[0])));
+        assert!(!pred(Any, &[0]).fires(&set(&[5])));
+    }
+
+    #[test]
+    fn split_keeps_old_rep_subclass_id_and_issues_the_other() {
+        // Round 1: {2,3} and {2,4} share a class (only ANY{2} distinguishes), rep = {2,3}, id 1.
+        let p1 = [policy(pred(Any, &[2]), pred(Any, &[9]))];
+        let r1 = allocate(&Registry::new(), &sets(&[&[2, 3], &[2, 4]]), &p1, 0).unwrap();
+        assert_eq!(r1.id_of(&set(&[2, 3]), &p1), Some(1));
+        assert_eq!(r1.id_of(&set(&[2, 4]), &p1), Some(1));
+
+        // Round 2: add EXACT{2,3}, which fires only on {2,3}, so the class splits.
+        let p2 = [
+            policy(pred(Any, &[2]), pred(Any, &[9])),
+            policy(pred(Exact, &[2, 3]), pred(Any, &[9])),
+        ];
+        let r2 = allocate(&r1, &sets(&[&[2, 3], &[2, 4]]), &p2, 0).unwrap();
+
+        // {2,3} carries the old representative, so it keeps id 1, and {2,4} issues a fresh id 2.
+        assert_eq!(
+            r2.id_of(&set(&[2, 3]), &p2),
+            Some(1),
+            "old-rep subclass keeps the id"
+        );
+        assert_eq!(
+            r2.id_of(&set(&[2, 4]), &p2),
+            Some(2),
+            "the other subclass issues fresh"
+        );
+        assert_eq!(r2.next(), 3);
+    }
+
+    #[test]
+    fn merge_retires_the_fused_id_and_prefers_fresh() {
+        // Start split: {2,3}=id1, {2,4}=id2 (EXACT{2,3} distinguishes them).
+        let p_split = [
+            policy(pred(Any, &[2]), pred(Any, &[9])),
+            policy(pred(Exact, &[2, 3]), pred(Any, &[9])),
+        ];
+        let r_split = allocate(&Registry::new(), &sets(&[&[2, 3], &[2, 4]]), &p_split, 0).unwrap();
+        assert_eq!(r_split.id_of(&set(&[2, 3]), &p_split), Some(1));
+        assert_eq!(r_split.id_of(&set(&[2, 4]), &p_split), Some(2));
+        assert_eq!(r_split.next(), 3);
+
+        // Drop the distinguisher, so {2,3} and {2,4} merge. The lex-min rep ({2,3}, id 1) survives
+        // and id 2 retires, gone from the live classes, quarantined, and next stays 3, so it is not
+        // reissued while fresh space remains.
+        let p_merge = [policy(pred(Any, &[2]), pred(Any, &[9]))];
+        let r_merge = allocate(&r_split, &sets(&[&[2, 3], &[2, 4]]), &p_merge, 0).unwrap();
+        assert_eq!(r_merge.id_of(&set(&[2, 3]), &p_merge), Some(1));
+        assert_eq!(
+            r_merge.id_of(&set(&[2, 4]), &p_merge),
+            Some(1),
+            "merged into the survivor"
+        );
+        assert!(!r_merge.classes().contains_key(&2), "id 2 is retired");
+        assert_eq!(r_merge.retired().get(&2), Some(&0), "id 2 is quarantined");
+        assert_eq!(r_merge.next(), 3, "the counter never rewinds");
+
+        // A genuinely new class (here {9}, matched as a destination) issues id 3, not the retired 2.
+        let r_next = allocate(&r_merge, &sets(&[&[2, 3], &[2, 4], &[9]]), &p_merge, 0).unwrap();
+        assert_eq!(
+            r_next.id_of(&set(&[9]), &p_merge),
+            Some(3),
+            "new class skips the tombstone"
+        );
+    }
+
+    #[test]
+    fn stable_ids_across_an_unrelated_edit() {
+        // Adding a class must not disturb the id of an existing, unaffected one.
+        let policies = [
+            policy(pred(Any, &[2]), pred(Any, &[9])),
+            policy(pred(Any, &[5]), pred(Any, &[9])),
+        ];
+        let r1 = allocate(&Registry::new(), &sets(&[&[2, 3]]), &policies, 0).unwrap();
+        let id_23 = r1.id_of(&set(&[2, 3]), &policies);
+        let r2 = allocate(&r1, &sets(&[&[2, 3], &[5]]), &policies, 0).unwrap();
+        assert_eq!(
+            r2.id_of(&set(&[2, 3]), &policies),
+            id_23,
+            "untouched class keeps its id"
+        );
+    }
+
+    #[test]
+    fn registry_round_trips_through_string_keyed_json() {
+        // mimics the running config the Perl side reads back, where object keys are strings. A
+        // registry without a retired key (an older running config) reads as an empty quarantine.
+        let reg: Registry = serde_json::from_str(r#"{"next":2,"classes":{"1":[1,2]}}"#)
+            .expect("deserialize string-keyed classes");
+        assert_eq!(reg.next(), 2);
+        assert_eq!(reg.classes().get(&1), Some(&set(&[1, 2])));
+        assert!(reg.retired().is_empty());
+
+        let out = serde_json::to_string(&reg).expect("serialize");
+        assert!(
+            out.contains("\"classes\":{\"1\":"),
+            "classes uses string keys: {out}"
+        );
+        let back: Registry = serde_json::from_str(&out).expect("re-deserialize");
+        assert_eq!(back, reg);
+
+        // the quarantine round-trips with string keys as well
+        let reg: Registry =
+            serde_json::from_str(r#"{"next":3,"classes":{"1":[1,2]},"retired":{"2":1234}}"#)
+                .expect("deserialize string-keyed retired");
+        assert_eq!(reg.retired().get(&2), Some(&1234));
+        let out = serde_json::to_string(&reg).expect("serialize");
+        assert!(
+            out.contains("\"retired\":{\"2\":1234}"),
+            "retired uses string keys: {out}"
+        );
+    }
+
+    /// A registry whose fresh space is spent: every id except `free_id` is live, `retired_id` is
+    /// quarantined since `retired_at`, and `free_id` is neither. `next` is 0 (fresh exhausted).
+    fn exhausted_registry(retired_id: u16, retired_at: u64, free_id: u16) -> Registry {
+        let classes: BTreeMap<u16, GroupSet> = (1..=u16::MAX)
+            .filter(|&id| id != retired_id && id != free_id)
+            .map(|id| (id, set(&[id])))
+            .collect();
+        Registry::from_parts(0, classes, BTreeMap::from([(retired_id, retired_at)]))
+    }
+
+    #[test]
+    fn reclaims_a_retired_id_only_after_its_quarantine() {
+        // {2,5} fires both src predicates, no singleton does, so no previous class claims it and
+        // it needs to issue with no fresh space left.
+        let policies = [
+            policy(pred(Any, &[2]), pred(Any, &[9])),
+            policy(pred(Any, &[5]), pred(Any, &[9])),
+        ];
+        let retired_at = 1_000;
+        // 7 is retired, 11 is plain free space (never live, out of every bookkeeping)
+        let prev = exhausted_registry(7, retired_at, 11);
+
+        // one number is free outright, so the first issued id takes it regardless of the quarantine
+        let reg = allocate(&prev, &sets(&[&[2, 5]]), &policies, retired_at + 1).unwrap();
+        assert_eq!(reg.id_of(&set(&[2, 5]), &policies), Some(11));
+
+        // with 11 also live, the only candidate is the quarantined 7, and inside the quarantine
+        // issuing must fail rather than rebind a number packets may still carry
+        let mut prev = exhausted_registry(7, retired_at, 11);
+        prev.classes.insert(11, set(&[11]));
+        let before = retired_at + RETIRE_QUARANTINE_SECS - 1;
+        let err = allocate(&prev, &sets(&[&[2, 5]]), &policies, before).unwrap_err();
+        assert!(err.to_string().contains("exhausted"), "got: {err:#}");
+
+        // once the quarantine has passed, 7 is reclaimed and leaves the bookkeeping
+        let after = retired_at + RETIRE_QUARANTINE_SECS;
+        let reg = allocate(&prev, &sets(&[&[2, 5]]), &policies, after).unwrap();
+        assert_eq!(reg.id_of(&set(&[2, 5]), &policies), Some(7));
+        assert!(!reg.retired().contains_key(&7));
+    }
+
+    #[test]
+    fn prefers_fresh_ids_over_reclaimable_ones() {
+        // id 3 was retired ages ago, yet it issues the fresh 5, since reuse is a last resort, and
+        // the long-expired tombstone drops out of the quarantine bookkeeping.
+        let policies = [policy(pred(Any, &[2]), pred(Any, &[9]))];
+        let prev = Registry::from_parts(5, BTreeMap::new(), BTreeMap::from([(3, 0)]));
+        let reg = allocate(
+            &prev,
+            &sets(&[&[2]]),
+            &policies,
+            RETIRE_QUARANTINE_SECS * 10,
+        )
+        .unwrap();
+        assert_eq!(reg.id_of(&set(&[2]), &policies), Some(5));
+        assert_eq!(reg.next(), 6);
+        assert!(reg.retired().is_empty());
+    }
+}
diff --git a/proxmox-ve-config/src/sdn/microseg/mod.rs b/proxmox-ve-config/src/sdn/microseg/mod.rs
new file mode 100644
index 0000000..5217cb5
--- /dev/null
+++ b/proxmox-ve-config/src/sdn/microseg/mod.rs
@@ -0,0 +1,9 @@
+//! Microseg subsystem of the SDN config.
+//!
+//! Microseg is an identity-based stateless firewall. Each managed guest NIC is assigned a *set* of
+//! numeric *groups*. A per-host agent compiles those sets into wire identities (see [`identity`])
+//! and admin-defined policy *rules* into a `(src_id, dst_id) -> allow|deny` map. A rule matches
+//! traffic by a `src` and a `dst` *predicate*, each a match kind (`any`/`all`/`exact`) over a list
+//! of groups.
+
+pub mod identity;
diff --git a/proxmox-ve-config/src/sdn/mod.rs b/proxmox-ve-config/src/sdn/mod.rs
index 2133396..f647e87 100644
--- a/proxmox-ve-config/src/sdn/mod.rs
+++ b/proxmox-ve-config/src/sdn/mod.rs
@@ -1,6 +1,7 @@
 pub mod config;
 pub mod fabric;
 pub mod ipam;
+pub mod microseg;
 pub mod prefix_list;
 pub mod route_map;
 pub mod wireguard;
-- 
2.47.3