fix(route-steering): incrementally advance owned reconciliation

crates/opc-route-steering/Cargo.toml

@@ -17,7 +17,7 @@ publish = false
 opc-linux-route-sys = { path = "../opc-linux-route-sys", version = "0.2.0" }
 async-trait = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["rt", "time"] }
+tokio = { workspace = true, features = ["rt", "time", "sync"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

crates/opc-route-steering/README.md

@@ -244,11 +244,25 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
   must durably reconstruct the complete desired set before reconciling; an
   empty desired set intentionally garbage-collects every representable owned
   object in that scope. The backend does not persist product intent.
-- Every Linux read, mutation, and convergence operation acquires one
-  clone-shared lock inside its blocking worker. A pair holds the lock once
+- `LinuxRouteSteeringBackend::plan_owned_route_rules` creates an opaque,
+  backend-bound reconciliation receipt. Advance it with
+  `reconcile_owned_route_rules_step`; every call performs exactly one cursor
+  poll, entry classification/canonical insertion, ordered merge comparison, or
+  mutation/ACK/verification unit. A returned state is quiescent: it holds no
+  shared operation lock and leaves no worker, poll, or mutation in flight.
+  Plans bind the backend identity, scope, desired state, authoritative
+  baseline, and generation. `Superseded` (another exact or legacy mutation)
+  and `Indeterminate` (possibly transmitted without authoritative proof) are
+  terminal and must be discarded, never replayed. The legacy complete-scope
+  API retains its existing serialized authoritative behavior.
+- Legacy Linux reads, mutations, and complete-scope convergence acquire one
+  clone-shared lock inside their blocking worker. A pair holds the lock once
   through post-install verification and rollback. If its async waiter is
   cancelled, the worker retains the lock and completes; the caller must retry
-  to obtain the resulting typed state.
+  to obtain the resulting typed state. Stepped-plan calls are different: each
+  call executes its one bounded unit inline, releases the shared lock before
+  returning, and never leaves detached polling or mutation work after the
+  future is dropped.
 - A Linux mutation is counted as acknowledged only after exactly one matching
   zero-error `NLMSG_ERROR` ACK. Empty or `NOOP`-only datagrams do not complete
   the operation; `DONE`, arbitrary payload messages, duplicate ACKs, timeout,

crates/opc-route-steering/src/collection.rs

@@ -2,6 +2,7 @@
 
 use std::collections::BTreeMap;
 use std::fmt;
+use std::hash::{Hash, Hasher};
 use std::net::IpAddr;
 
 use crate::error::RouteSteeringError;
@@ -185,6 +186,7 @@ pub struct OwnedRouteRuleSet {
     scope: OwnedRouteRuleScope,
     routes: Vec<RouteRequest>,
     rules: Vec<RuleRequest>,
+    canonical_digest: u64,
 }
 
 impl OwnedRouteRuleSet {
@@ -208,10 +210,14 @@ impl OwnedRouteRuleSet {
             MAX_OWNED_ROUTE_COLLECTION_ENTRIES,
             MAX_OWNED_RULE_COLLECTION_ENTRIES,
         )
-        .map(|(routes, rules)| Self {
-            scope,
-            routes,
-            rules,
+        .map(|(routes, rules)| {
+            let canonical_digest = canonical_collection_digest(scope, &routes, &rules);
+            Self {
+                scope,
+                routes,
+                rules,
+                canonical_digest,
+            }
         })
     }
 
@@ -233,13 +239,69 @@ impl OwnedRouteRuleSet {
         &self.rules
     }
 
+    /// Deterministic digest of the already validated canonical desired set.
+    /// It is computed during construction, never logged, and lets opaque
+    /// backend plans bind to their exact desired authority without rescanning
+    /// a large collection at plan creation.
+    #[must_use]
+    pub(crate) const fn canonical_digest(&self) -> u64 {
+        self.canonical_digest
+    }
+
     /// Consume the set into its scope and canonical route and rule vectors.
     #[must_use]
     pub fn into_parts(self) -> (OwnedRouteRuleScope, Vec<RouteRequest>, Vec<RuleRequest>) {
         (self.scope, self.routes, self.rules)
     }
 }
 
+#[derive(Default)]
+struct CanonicalDigestHasher(u64);
+
+impl CanonicalDigestHasher {
+    const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
+    const PRIME: u64 = 0x0000_0100_0000_01b3;
+
+    fn with_domain(domain: u8) -> Self {
+        let mut hasher = Self(Self::OFFSET);
+        hasher.write_u8(domain);
+        hasher
+    }
+}
+
+impl Hasher for CanonicalDigestHasher {
+    fn finish(&self) -> u64 {
+        self.0
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        for byte in bytes {
+            self.0 ^= u64::from(*byte);
+            self.0 = self.0.wrapping_mul(Self::PRIME);
+        }
+    }
+}
+
+fn canonical_collection_digest(
+    scope: OwnedRouteRuleScope,
+    routes: &[RouteRequest],
+    rules: &[RuleRequest],
+) -> u64 {
+    let mut hasher = CanonicalDigestHasher::with_domain(1);
+    scope.hash(&mut hasher);
+    routes.len().hash(&mut hasher);
+    for route in routes {
+        2_u8.hash(&mut hasher);
+        route.hash(&mut hasher);
+    }
+    rules.len().hash(&mut hasher);
+    for rule in rules {
+        3_u8.hash(&mut hasher);
+        rule.hash(&mut hasher);
+    }
+    hasher.finish()
+}
+
 impl fmt::Debug for OwnedRouteRuleSet {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("OwnedRouteRuleSet")
@@ -305,6 +367,25 @@ impl OwnedRouteRuleSnapshot {
         })
     }
 
+    /// Build a snapshot from the stepped Linux collector.
+    ///
+    /// The collector has already validated scope membership, canonical route
+    /// keys, exact duplicates, and source-range sibling disjointness while it
+    /// consumed each netlink object.  Its ordered iterators materialize these
+    /// vectors one item per public advance, so re-running the normal
+    /// whole-collection validator here would defeat the bounded-step contract.
+    pub(crate) fn from_incremental_collector(
+        scope: OwnedRouteRuleScope,
+        routes: Vec<RouteRequest>,
+        rules: Vec<RuleRequest>,
+    ) -> Self {
+        Self {
+            scope,
+            routes,
+            rules,
+        }
+    }
+
     /// Exclusive-writer scope represented by this snapshot.
     #[must_use]
     pub const fn scope(&self) -> OwnedRouteRuleScope {

crates/opc-route-steering/src/lib.rs

@@ -49,7 +49,8 @@ pub use collection::{
 pub use error::{RouteSteeringError, RouteSteeringFailureClass};
 pub use linux::{
     LinuxOwnedRouteRuleCollectionLimits, LinuxRouteReadbackLimits, LinuxRouteSteeringBackend,
-    LinuxRouteSteeringBackendConfig, LinuxRuleProtocolCapability, LINUX_ROUTE_STEERING_PROTOCOL,
+    LinuxRouteSteeringBackendConfig, LinuxRouteSteeringUrgentIntent, LinuxRuleProtocolCapability,
+    OwnedRouteRuleReconcilePlan, OwnedRouteRuleReconcileStep, LINUX_ROUTE_STEERING_PROTOCOL,
 };
 pub use mock::{MockFailurePoint, MockObservation, MockOperation, MockRouteSteeringBackend};
 pub use model::{