Nested unfold whose object takes a higher-layer product leaves that layer unflushed #955

// job -> spill --split_a--> a --split_b--> b
//
// split_b's object takes the a-layer product and the spill-layer product.
// With two_inputs: false it takes only the a-layer product.
#include "phlex/core/product_selector.hpp"
#include "phlex/model/data_cell_index.hpp"
#include "phlex/module.hpp"
#include "phlex/source.hpp"

#include <spdlog/spdlog.h>
#include <utility>

namespace {
// Unfolds into k children.
struct Split {
  explicit Split(int k) : k{k} {}
  int initial_value() const { return 0; }
  int k;
};
// Same, but also receives a product from a higher layer.
struct SplitWithSpill {
  SplitWithSpill(int k, int /*spill*/) : k{k} {}
  int initial_value() const { return 0; }
  int k;
};
} // namespace

PHLEX_REGISTER_PROVIDERS(s, config)
{
  s.provide("provide_n", [](phlex::data_cell_index const&) { return 2; })
    .output_product("src", phlex::experimental::identifier{"n"},
                    phlex::experimental::identifier{"spill"});
}

PHLEX_REGISTER_ALGORITHMS(m, config)
{
  using phlex::product_selector;
  using id = phlex::experimental::identifier;
  product_selector const spill_n{.creator = "src", .layer = id{"spill"}, .suffix = "n"};
  product_selector const a_n{.creator = "split_a", .layer = id{"a"}, .suffix = "n"};
  product_selector const b_n{.creator = "split_b", .layer = id{"b"}, .suffix = "n"};

  // Each spill -> 2 a cells, each carrying 1.
  m.unfold<Split>(
     "split_a", [](Split const& s, int i) { return i < s.k; },
     [](Split const&, int i) { return std::make_pair(i + 1, 1); }, "a",
     phlex::concurrency::unlimited)
    .input_family(spill_n)
    .output_product_suffixes("n");

  // Each a cell -> 1 b cell.
  if (config.get<bool>("two_inputs", true)) {
    m.unfold<SplitWithSpill>(
       "split_b", [](SplitWithSpill const& s, int i) { return i < s.k; },
       [](SplitWithSpill const&, int i) { return std::make_pair(i + 1, 1); }, "b",
       phlex::concurrency::unlimited)
      .input_family(a_n, spill_n)
      .output_product_suffixes("n");
  } else {
    m.unfold<Split>(
       "split_b", [](Split const& s, int i) { return i < s.k; },
       [](Split const&, int i) { return std::make_pair(i + 1, 1); }, "b",
       phlex::concurrency::unlimited)
      .input_family(a_n)
      .output_product_suffixes("n");
  }

  // Count b cells per spill.
  m.fold(
     "count", [](int& c, int) { ++c; }, phlex::concurrency::serial, "spill")
    .input_family(b_n)
    .output_product_suffixes("count");

  m.observe(
     "print", [](int c) { spdlog::info("count = {}", c); }, phlex::concurrency::serial)
    .input_family(product_selector{.creator = "count", .layer = id{"spill"}, .suffix = "count"});
}