Fixed and lossy equal-balanced placement, reverse SPF (0.9.0)

CHANGELOG.md

@@ -5,10 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [0.9.0] - 2026-09-13
+
+### Added
+
+- **Shortest Paths**: `shortest_paths_to(dst, ...)` / `Algorithms.spf_to` runs Dijkstra over the in-adjacency and returns distances *to* one destination plus a forward-oriented `PredDAG` valid for placement from any node toward it (the union of the per-source shortest-path DAGs). `fanout_edges` adds edges regardless of cost, for an origin whose first hop is a traffic split rather than a routing decision, such as a pseudo source whose demand originates evenly at every attached real source; a fan-out edge must leave a node with no incoming DAG entry, which keeps the result acyclic.
+- **Flow Placement**: two equal-balanced modes for hop-by-hop ECMP with a load-blind forwarding table. `EQUAL_BALANCED_FIXED` takes the split set from the DAG edges with *capacity* rather than residual and keeps the single global admission scale, so a member filled since the DAG was computed drives the scale to 0 (lossless hash-ECMP admission; the existing `EQUAL_BALANCED` drops such members from the split, which is progressive/TE behaviour). `EQUAL_BALANCED_LOSSY` splits over the same set without a scale: each edge carries `min(share, residual)`, the excess is dropped, deficits propagate downstream, and the placed amount is what reaches the destination. `FlowGraph.place_with_drops` returns the dropped volume per edge; `FlowState::place_on_dag` gained an optional `drops` collector. On a 100/10 parallel pair offered 100 units, `EQUAL_BALANCED` and `EQUAL_BALANCED_FIXED` admit 20 and `EQUAL_BALANCED_LOSSY` delivers 60. `FlowPolicy` treats the new modes as equal-balanced except that the lossy mode skips the equalizing rebalance, so with pinned routes each LSP carries what fits and `placed` is the delivered total.
 
 ### Changed
 
+- **Flow Policy**: a cost-only policy (`require_capacity=false`) no longer passes the residual to SPF when an equal-balanced per-flow target is set. A residual forces capacity-aware edge selection, so such policies silently routed around saturated edges instead of following costs; they now route on cost alone, and the per-flow target still bounds how much each flow requests.
+
 - **Flow Policy**: `get_path_bundle` now memoizes raw SPF results keyed by the exact inputs that can vary per call (src, dst, residual content, residual-awareness). EqualBalanced placement and rebalance rounds re-request bundles against residual state that repeats -- 94% of SPF calls in a measured place/rebalance cycle were exact input repeats, largely remove+place round-trips restoring identical bytes -- and those calls are now elided. Matching is exact (FlowGraph state stamp fast path, full residual `memcmp` content path), so all outputs are bit-identical; a corpus hash over max-flow, SPF, KSP and policy outputs is unchanged. Measured on place/rebalance churn: 31-70% faster; single EqualBalanced placement: 6-26% faster; Proportional mode skips the memo (it measured 0% repeat inputs) and is unaffected. `FlowGraph` gained an internal monotonic state stamp to support this; no public API change.
 
 ## [0.8.0] - 2026-08-24

bindings/python/module.cpp

@@ -110,7 +110,9 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) {
 
   py::enum_<FlowPlacement>(m, "FlowPlacement")
       .value("PROPORTIONAL", FlowPlacement::Proportional)
-      .value("EQUAL_BALANCED", FlowPlacement::EqualBalanced);
+      .value("EQUAL_BALANCED", FlowPlacement::EqualBalanced)
+      .value("EQUAL_BALANCED_FIXED", FlowPlacement::EqualBalancedFixed)
+      .value("EQUAL_BALANCED_LOSSY", FlowPlacement::EqualBalancedLossy);
 
   py::class_<StrictMultiDiGraph>(m, "StrictMultiDiGraph")
       .def_static(
@@ -218,6 +220,56 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) {
           throw py::value_error("dtype must be 'float64' or 'int64'");
         }
       }, py::arg("graph"), py::arg("src"), py::arg("dst") = py::none(), py::kw_only(), py::arg("selection") = py::none(), py::arg("residual") = py::none(), py::arg("node_mask") = py::none(), py::arg("edge_mask") = py::none(), py::arg("multipath") = true, py::arg("dtype") = "float64")
+      .def("spf_to", [](const Algorithms& algs, const PyGraph& pg, std::int32_t dst,
+                          py::object selection_obj, py::object residual_obj,
+                          py::object node_mask, py::object edge_mask, bool multipath,
+                          py::object fanout_obj, std::string dtype) -> py::tuple {
+        if (dst < 0 || dst >= pg.num_nodes) throw py::value_error("dst out of range");
+        SpfToOptions opts; if (!selection_obj.is_none()) opts.selection = py::cast<EdgeSelection>(selection_obj);
+        std::vector<double> residual_vec;
+        if (!residual_obj.is_none()) {
+          auto arr = py::cast<py::array>(residual_obj);
+          if (!(arr.flags() & py::array::c_style)) throw py::type_error("residual must be C-contiguous (np.ascontiguousarray)");
+          auto buf = arr.request();
+          if (buf.ndim != 1 || buf.format != py::format_descriptor<double>::format()) throw py::type_error("residual must be 1-D float64");
+          if (static_cast<std::int32_t>(buf.shape[0]) != pg.num_edges) {
+            throw py::type_error("residual length must equal " + std::to_string(pg.num_edges));
+          }
+          residual_vec.resize(static_cast<std::size_t>(buf.shape[0]));
+          std::memcpy(residual_vec.data(), buf.ptr, residual_vec.size()*sizeof(double));
+          opts.residual = std::span<const double>(residual_vec.data(), residual_vec.size());
+        }
+        std::vector<EdgeId> fanout_vec;
+        if (!fanout_obj.is_none()) {
+          for (auto item : py::iterable(fanout_obj)) {
+            auto e = py::cast<std::int64_t>(item);
+            if (e < 0 || e >= pg.num_edges) throw py::value_error("fanout edge id out of range");
+            fanout_vec.push_back(static_cast<EdgeId>(e));
+          }
+          opts.fanout_edges = std::span<const EdgeId>(fanout_vec.data(), fanout_vec.size());
+        }
+        auto node_bs = to_bool_span_from_numpy(node_mask, static_cast<std::size_t>(pg.num_nodes), "node_mask");
+        auto edge_bs = to_bool_span_from_numpy(edge_mask, static_cast<std::size_t>(pg.num_edges), "edge_mask");
+        opts.node_mask = node_bs.view;
+        opts.edge_mask = edge_bs.view;
+        opts.multipath = multipath;
+        py::gil_scoped_release rel; auto res = algs.spf_to(pg.handle, dst, opts); py::gil_scoped_acquire acq;
+        auto maxc = std::numeric_limits<Cost>::max();
+        if (dtype == "int64") {
+          py::array_t<std::int64_t> dist_arr(res.first.size());
+          auto* out = dist_arr.mutable_data();
+          for (std::size_t i=0;i<res.first.size();++i) out[i] = (res.first[i]==maxc) ? static_cast<std::int64_t>(maxc) : static_cast<std::int64_t>(res.first[i]);
+          return py::make_tuple(std::move(dist_arr), res.second);
+        } else if (dtype == "float64") {
+          py::array_t<double> dist_arr(res.first.size());
+          auto* out = dist_arr.mutable_data();
+          for (std::size_t i=0;i<res.first.size();++i) out[i] = (res.first[i]==maxc) ? std::numeric_limits<double>::infinity() : static_cast<double>(res.first[i]);
+          return py::make_tuple(std::move(dist_arr), res.second);
+        } else {
+          throw py::value_error("dtype must be 'float64' or 'int64'");
+        }
+      }, py::arg("graph"), py::arg("dst"), py::kw_only(), py::arg("selection") = py::none(), py::arg("residual") = py::none(), py::arg("node_mask") = py::none(), py::arg("edge_mask") = py::none(), py::arg("multipath") = true, py::arg("fanout_edges") = py::none(), py::arg("dtype") = "float64",
+         "Shortest paths from every node to dst (reverse SPF). Returns (distances_to_dst, dag); the DAG is forward-oriented and valid for placement from any node toward dst. fanout_edges are added to the DAG regardless of cost (see shortest_paths.hpp).")
       .def("ksp", [](const Algorithms& algs, const PyGraph& pg, std::int32_t src, std::int32_t dst,
                        int k, py::object max_cost_factor, bool unique, py::object node_mask, py::object edge_mask, std::string dtype){
         if (src < 0 || src >= pg.num_nodes || dst < 0 || dst >= pg.num_nodes) throw py::value_error("src/dst out of range");
@@ -489,6 +541,14 @@ PYBIND11_MODULE(_netgraph_core, m, py::mod_gil_not_used()) {
       })
       .def_property_readonly("graph", [](const FlowGraph& fg){ return &fg.graph(); }, py::return_value_policy::reference_internal)
       .def("place", [](FlowGraph& fg, const FlowIndex& idx, std::int32_t src, std::int32_t dst, const PredDAG& dag, double amount, FlowPlacement placement){ py::gil_scoped_release rel; auto placed = fg.place(idx, src, dst, dag, amount, placement); py::gil_scoped_acquire acq; return placed; }, py::arg("index"), py::arg("src"), py::arg("dst"), py::arg("dag"), py::arg("amount"), py::arg("flow_placement") = FlowPlacement::Proportional)
+      .def("place_with_drops", [](FlowGraph& fg, const FlowIndex& idx, std::int32_t src, std::int32_t dst, const PredDAG& dag, double amount, FlowPlacement placement){
+        std::vector<std::pair<EdgeId, Flow>> drops;
+        double placed;
+        { py::gil_scoped_release rel; placed = fg.place(idx, src, dst, dag, amount, placement, &drops); }
+        py::list out; for (auto const& pr : drops) out.append(py::make_tuple(pr.first, pr.second));
+        return py::make_tuple(placed, out);
+      }, py::arg("index"), py::arg("src"), py::arg("dst"), py::arg("dag"), py::arg("amount"), py::arg("flow_placement") = FlowPlacement::EqualBalancedLossy,
+         "Like place(), but also returns the per-edge dropped volume as a list of (edge_id, dropped) pairs. Only EQUAL_BALANCED_LOSSY drops flow; other placements return an empty list.")
       .def("remove", [](FlowGraph& fg, const FlowIndex& idx){ py::gil_scoped_release rel; fg.remove(idx); py::gil_scoped_acquire acq; })
       .def("remove_by_class", [](FlowGraph& fg, std::int32_t cls){ py::gil_scoped_release rel; fg.remove_by_class(cls); py::gil_scoped_acquire acq; })
       .def("reset", [](FlowGraph& fg){ py::gil_scoped_release rel; fg.reset(); py::gil_scoped_acquire acq; })

include/netgraph/core/algorithms.hpp

@@ -21,6 +21,11 @@ class Algorithms {
     return backend_->spf(gh, src, opts);
   }
 
+  [[nodiscard]] std::pair<std::vector<Cost>, PredDAG>
+  spf_to(const GraphHandle& gh, NodeId dst, const SpfToOptions& opts) const {
+    return backend_->spf_to(gh, dst, opts);
+  }
+
   [[nodiscard]] std::vector<std::pair<std::vector<Cost>, PredDAG>>
   ksp(const GraphHandle& gh, NodeId src, NodeId dst, const KspOptions& opts) const {
     return backend_->ksp(gh, src, dst, opts);

include/netgraph/core/backend.hpp

@@ -55,6 +55,12 @@ class Backend {
   [[nodiscard]] virtual std::pair<std::vector<Cost>, PredDAG> spf(
       const GraphHandle& gh, NodeId src, const SpfOptions& opts) = 0;
 
+  // Computes shortest paths from every node to dst (reverse SPF); see
+  // shortest_paths_to() in shortest_paths.hpp. Returns distances to dst and a
+  // forward-oriented predecessor DAG valid for placement from any node toward dst.
+  [[nodiscard]] virtual std::pair<std::vector<Cost>, PredDAG> spf_to(
+      const GraphHandle& gh, NodeId dst, const SpfToOptions& opts) = 0;
+
   // Computes maximum flow between a source and destination node.
   //
   // Arguments:

include/netgraph/core/flow_graph.hpp

@@ -40,10 +40,15 @@ class FlowGraph {
   // Access underlying graph (const)
   [[nodiscard]] const StrictMultiDiGraph& graph() const noexcept { return *g_; }
 
-// Apply placement and record per-edge allocations for this flow. Returns placed amount.
+  // Apply placement and record per-edge allocations for this flow. Returns placed
+  // amount. `drops`, when given, receives the per-edge dropped volume of an
+  // EqualBalancedLossy placement (see FlowState::place_on_dag); other placements
+  // leave it untouched. Dropped volume is not part of the ledger: it never
+  // occupied an edge, so removing the flow reverts only what was carried.
   [[nodiscard]] Flow place(const FlowIndex& idx, NodeId src, NodeId dst,
              const PredDAG& dag, Flow amount,
-             FlowPlacement placement);
+             FlowPlacement placement,
+             std::vector<std::pair<EdgeId, Flow>>* drops = nullptr);
 
   // Remove a specific flow, reverting its edge allocations from the ledger.
   void remove(const FlowIndex& idx);

include/netgraph/core/flow_policy.hpp

@@ -146,9 +146,12 @@ class FlowPolicy {
   //
   // With static paths configured the policy neither creates additional flows nor
   // reoptimizes: max_path_cost, max_path_cost_factor, min_flow_count and
-  // reoptimize_flows_on_each_placement are inert. EqualBalanced spreads over the
-  // usable (up) bundles only. flow_count() reports the usable count U; the
-  // supplied count N is the caller's, so down LSPs = N - flow_count().
+  // reoptimize_flows_on_each_placement are inert. EqualBalanced and
+  // EqualBalancedFixed spread over the usable (up) bundles only and equalize the
+  // carried volume per bundle; EqualBalancedLossy offers each up bundle an equal
+  // share and lets it carry what fits (no equalizing rebalance), so placed is the
+  // delivered total. flow_count() reports the usable count U; the supplied count
+  // N is the caller's, so down LSPs = N - flow_count().
   //
   // Throws std::invalid_argument if bundles is empty, the policy already holds
   // flows (remove_demand() first), shortest_path=true is configured (single-

include/netgraph/core/flow_state.hpp

@@ -53,12 +53,25 @@ class FlowState {
   //    saturates. Re-invoking this on the updated residuals changes the effective
   //    next-hop set (progressive traffic-engineering behavior) and is outside
   //    "single-pass ECMP admission".
+  //
+  // EqualBalancedFixed uses the same global scale but takes the split set from
+  // the DAG edges with capacity rather than with residual, so a member that has
+  // been saturated since the DAG was built yields scale 0 and nothing is placed.
+  //
+  // EqualBalancedLossy splits over the same capacity-based set without scaling:
+  // each edge carries min(share, residual), the excess is dropped, and the
+  // returned amount is what reaches dst. When `drops` is given it receives one
+  // (edge, dropped) entry per edge that dropped flow. Drops are recorded only
+  // for finite requested_flow; an infinite request fills every source edge and
+  // records no drops. Other placements never write to `drops`.
   [[nodiscard]] Flow place_on_dag(NodeId src, NodeId dst,
                     const PredDAG& dag,
                     Flow requested_flow,
                     FlowPlacement placement,
                     // Optional trace collector to record per-edge allocations applied by this call
-                    std::vector<std::pair<EdgeId, Flow>>* trace = nullptr);
+                    std::vector<std::pair<EdgeId, Flow>>* trace = nullptr,
+                    // Optional collector for per-edge dropped volume (EqualBalancedLossy only)
+                    std::vector<std::pair<EdgeId, Flow>>* drops = nullptr);
 
   // Convenience: run repeated placements until exhaustion (or single tier when
   // shortest_path=true). Returns total placed flow. Uses internal residual.

include/netgraph/core/options.hpp

@@ -17,6 +17,17 @@ struct SpfOptions {
   std::span<const bool> edge_mask {};
 };
 
+// Options for shortest_paths_to (reverse SPF toward one destination). See
+// shortest_paths.hpp for the semantics of fanout_edges.
+struct SpfToOptions {
+  bool multipath { true };
+  EdgeSelection selection {};
+  std::span<const Cap> residual {};
+  std::span<const bool> node_mask {};
+  std::span<const bool> edge_mask {};
+  std::span<const EdgeId> fanout_edges {};
+};
+
 struct KspOptions {
   int k { 1 };
   std::optional<double> max_cost_factor {};

include/netgraph/core/shortest_paths.hpp

@@ -93,3 +93,39 @@ resolve_to_paths(const PredDAG& dag, NodeId src, NodeId dst,
                  std::optional<std::int64_t> max_paths = std::nullopt);
 
 } // namespace netgraph::core
+
+namespace netgraph::core {
+
+// Compute shortest paths *to* dst from every node (Dijkstra over the in-adjacency).
+// Returns (distances, dag) where distances[u] is the cost of a shortest u -> dst walk
+// (INT64_MAX if none) and dag is a forward-oriented PredDAG: parents[v] holds (u, e)
+// for every edge e = u -> v that lies on a shortest u -> dst walk. Unlike a DAG from
+// shortest_paths(), which is rooted at one source, this DAG is valid for placement
+// from *any* node toward dst, and every walk it contains from a node u to dst has
+// cost distances[u]. It is the union of the per-source shortest-path DAGs toward dst.
+//
+// Parameters mirror shortest_paths(): multipath keeps every equal-cost successor per
+// node (false keeps one, preferring higher bottleneck capacity toward dst);
+// selection, residual, node_mask and edge_mask apply as there. Zero-cost edges get
+// the same acyclicity guard (a successor is recorded only while its node is
+// unsettled).
+//
+// fanout_edges: edges to add to the DAG regardless of cost, modelling an origin
+// whose first hop is decided by a traffic split rather than by routing (e.g. a
+// pseudo source whose demand originates evenly at every attached real source).
+// Each edge u -> v is added as a parent entry of v when it passes the masks and
+// capacity gate and v has a finite distance; entries the SPF already recorded are
+// not duplicated, and distances[u] is set to cost(e) + distances[v] if u was
+// unreachable. Every fanout edge must leave a node that has no incoming DAG entry,
+// so the result stays acyclic; a violating edge throws std::invalid_argument, as
+// does an out-of-range edge id.
+[[nodiscard]] std::pair<std::vector<Cost>, PredDAG>
+shortest_paths_to(const StrictMultiDiGraph& g, NodeId dst,
+                  bool multipath,
+                  const EdgeSelection& selection,
+                  std::span<const Cap> residual = {},
+                  std::span<const bool> node_mask = {},
+                  std::span<const bool> edge_mask = {},
+                  std::span<const EdgeId> fanout_edges = {});
+
+} // namespace netgraph::core

include/netgraph/core/types.hpp

@@ -56,9 +56,28 @@ struct FlowIndexHash {
 //   out of scope for EqualBalanced.
 //
 // - Proportional may be used iteratively (e.g., for max-flow).
+//
+// - EqualBalanced builds its split set from the DAG edges that currently have
+//   residual, so re-invoking it on updated residuals shrinks the next-hop set
+//   (progressive behavior, used by place_max_flow and FlowPolicy fills).
+//
+// - EqualBalancedFixed builds its split set from the DAG edges that have
+//   *capacity* (the topology's next-hop set), and admits with the same single
+//   global scale computed from residual headroom. A member saturated since the
+//   DAG was built therefore blocks admission entirely (scale 0): this is
+//   lossless hash-ECMP admission with a forwarding table that does not react
+//   to load.
+//
+// - EqualBalancedLossy also splits over the topology's next-hop set, but does
+//   not scale: every edge carries min(share, residual) and the excess is
+//   dropped, deficits propagate downstream, and the placed amount is the volume
+//   that reaches dst. This is best-effort hash-ECMP forwarding; the dropped
+//   share per edge is available through the drop trace of place_on_dag.
 enum class FlowPlacement {
-  Proportional = 1,    // Distribute flow proportionally to residual capacity (like ECMP with weights)
-  EqualBalanced = 2    // Split equally per parallel edge on a fixed DAG (single-pass ECMP admission)
+  Proportional = 1,        // Distribute flow proportionally to residual capacity (like ECMP with weights)
+  EqualBalanced = 2,       // Split equally per parallel edge on a fixed DAG (single-pass ECMP admission)
+  EqualBalancedFixed = 3,  // Equal split over the topology next-hop set; a saturated member blocks admission
+  EqualBalancedLossy = 4   // Equal split over the topology next-hop set; excess over residual is dropped
 };
 
 // Tie-breaking rule when multiple equal-cost edges exist between the same (u,v) pair.