Use payment flow code for offer crossing (RIPD-1094):

Replace Taker.cpp with calls to the payment flow() code.

This change required a number of tweaks in the payment flow code.
These tweaks are conditionalized on whether or not offer crossing
is taking place.  The flag is explicitly passed as a parameter to
the flow code.

For testing, a class was added that identifies differences in the
contents of two PaymentSandboxes.  That code may be reusable in
the future.

None of the Taker offer crossing code is removed.  Both versions
of the code are co-resident to support an amendment cut-over.

The code that identifies differences between Taker and Flow offer
crossing is enabled by a feature.  That makes it easy to enable
or disable difference logging by changing the config file.  This
approach models what was done with the payment flow code.  The
differencing code should never be enabled on a production server.

Extensive offer crossing unit tests are added to examine and
verify the behavior of corner cases.  The tests are currently
configured to run against both Taker and Flow offer crossing.
This gives us confidence that most cases run identically and
some of the (few) differences in behavior are documented.

src/ripple/app/paths/impl/DirectStep.cpp

@@ -33,17 +33,17 @@
 
 namespace ripple {
 
-class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI>
+template <class TDerived>
+class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI<TDerived>>
 {
-  private:
+protected:
     AccountID src_;
     AccountID dst_;
     Currency currency_;
-    bool isLast_ = false;
 
     // Charge transfer fees when the prev step redeems
     Step const* const prevStep_ = nullptr;
-
+    bool const isLast_;
     beast::Journal j_;
 
     struct Cache
@@ -67,27 +67,33 @@ class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI>
 
     boost::optional<Cache> cache_;
 
+    // Compute the maximum value that can flow from src->dst at
+    // the best available quality.
+    // return: first element is max amount that can flow,
+    //         second is true if dst holds an iou from src.
+    std::pair<IOUAmount, bool>
+    maxPaymentFlow (
+        ReadView const& sb) const;
+
     // Returns srcQOut, dstQIn
     std::pair <std::uint32_t, std::uint32_t>
     qualities (
-        PaymentSandbox& sb,
+        ReadView const& sb,
         bool srcRedeems,
         bool fwd) const;
 
-  public:
+public:
     DirectStepI (
+        StrandContext const& ctx,
         AccountID const& src,
         AccountID const& dst,
-        Currency const& c,
-        Step const* prevStep,
-        bool isLast,
-        beast::Journal j)
-            :src_(src)
+        Currency const& c)
+            : src_(src)
             , dst_(dst)
             , currency_ (c)
-            , isLast_ (isLast)
-            , prevStep_ (prevStep)
-            , j_ (j)
+            , prevStep_ (ctx.prevStep)
+            , isLast_ (ctx.isLast)
+            , j_ (ctx.j)
     {}
 
     AccountID const& src () const
@@ -136,6 +142,9 @@ class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI>
     std::uint32_t
     lineQualityIn (ReadView const& v) const override;
 
+    boost::optional<Quality>
+    qualityUpperBound(ReadView const& v, bool& redeems) const override;
+
     std::pair<IOUAmount, IOUAmount>
     revImp (
         PaymentSandbox& sb,
@@ -178,7 +187,18 @@ class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI>
         return ! (lhs == rhs);
     }
 
-  private:
+protected:
+    std::string logStringImpl (char const* name) const
+    {
+        std::ostringstream ostr;
+        ostr <<
+            name << ": " <<
+            "\nSrc: " << src_ <<
+            "\nDst: " << dst_;
+        return ostr.str ();
+    }
+
+private:
     bool equal (Step const& rhs) const override
     {
         if (auto ds = dynamic_cast<DirectStepI const*> (&rhs))
@@ -187,62 +207,288 @@ class DirectStepI : public StepImp<IOUAmount, IOUAmount, DirectStepI>
         }
         return false;
     }
+};
+
+//------------------------------------------------------------------------------
+
+// Flow is used in two different circumstances for transferring funds:
+//  o Payments, and
+//  o Offer crossing.
+// The rules for handling funds in these two cases are almost, but not
+// quite, the same.
+
+// Payment DirectStep class (not offer crossing).
+class DirectIPaymentStep : public DirectStepI<DirectIPaymentStep>
+{
+public:
+    using DirectStepI<DirectIPaymentStep>::DirectStepI;
+    using DirectStepI<DirectIPaymentStep>::check;
+
+    bool verifyPrevStepRedeems (bool) const
+    {
+        // A payment doesn't care whether or not prevStepRedeems.
+        return true;
+    }
+
+    bool verifyDstQualityIn (std::uint32_t dstQIn) const
+    {
+        // Payments have no particular expectations for what dstQIn will be.
+        return true;
+    }
+
+    std::uint32_t
+    quality (ReadView const& sb,
+        // set true for quality in, false for quality out
+        bool qin) const;
+
+    // Compute the maximum value that can flow from src->dst at
+    // the best available quality.
+    // return: first element is max amount that can flow,
+    //         second is true if dst holds an iou from src.
+    std::pair<IOUAmount, bool>
+    maxFlow (ReadView const& sb, IOUAmount const& desired) const;
+
+    // Verify the consistency of the step.  These checks are specific to
+    // payments and assume that general checks were already performed.
+    TER
+    check (StrandContext const& ctx,
+        std::shared_ptr<const SLE> const& sleSrc) const;
 
     std::string logString () const override
     {
-        std::ostringstream ostr;
-        ostr <<
-            "DirectStepI: " <<
-            "\nSrc: " << src_ <<
-            "\nDst: " << dst_;
-        return ostr.str ();
+        return logStringImpl ("DirectIPaymentStep");
     }
 };
 
-// Compute the maximum value that can flow from src->dst at
-// the best available quality.
-// return: first element is max amount that can flow,
-//         second is if src redeems to dst
-static
+// Offer crossing DirectStep class (not a payment).
+class DirectIOfferCrossingStep : public DirectStepI<DirectIOfferCrossingStep>
+{
+public:
+    using DirectStepI<DirectIOfferCrossingStep>::DirectStepI;
+    using DirectStepI<DirectIOfferCrossingStep>::check;
+
+    bool verifyPrevStepRedeems (bool prevStepRedeems) const
+    {
+        // During offer crossing we rely on the fact that prevStepRedeems
+        // will *always* be false.  That's because:
+        //  o If there's a prevStep_, it will always be a BookStep.
+        //  o BookStep::redeems() aways returns false when offer crossing.
+        // An assert based on this return value will tell us if that
+        // behavior changes.
+        return !prevStepRedeems;
+    }
+
+    bool verifyDstQualityIn (std::uint32_t dstQIn) const
+    {
+        // Due to a couple of factors dstQIn is always QUALITY_ONE for
+        // offer crossing.  If that changes we need to know.
+        return dstQIn == QUALITY_ONE;
+    }
+
+    std::uint32_t
+    quality (ReadView const& sb,
+        // set true for quality in, false for quality out
+        bool qin) const;
+
+    // Compute the maximum value that can flow from src->dst at
+    // the best available quality.
+    // return: first element is max amount that can flow,
+    //         second is true if dst holds an iou from src.
+    std::pair<IOUAmount, bool>
+    maxFlow (ReadView const& sb, IOUAmount const& desired) const;
+
+    // Verify the consistency of the step.  These checks are specific to
+    // offer crossing and assume that general checks were already performed.
+    TER
+    check (StrandContext const& ctx,
+        std::shared_ptr<const SLE> const& sleSrc) const;
+
+
+    std::string logString () const override
+    {
+        return logStringImpl ("DirectIOfferCrossingStep");
+    }
+};
+
+//------------------------------------------------------------------------------
+
+std::uint32_t
+DirectIPaymentStep::quality (ReadView const& sb,
+    // set true for quality in, false for quality out
+    bool qin) const
+{
+    if (src_ == dst_)
+        return QUALITY_ONE;
+
+    auto const sle = sb.read (keylet::line (dst_, src_, currency_));
+
+    if (!sle)
+        return QUALITY_ONE;
+
+    auto const& field = [this, qin]() -> SF_U32 const&
+    {
+        if (qin)
+        {
+            // compute dst quality in
+            if (this->dst_ < this->src_)
+                return sfLowQualityIn;
+            else
+                return sfHighQualityIn;
+        }
+        else
+        {
+            // compute src quality out
+            if (this->src_ < this->dst_)
+                return sfLowQualityOut;
+            else
+                return sfHighQualityOut;
+        }
+    }();
+
+    if (! sle->isFieldPresent (field))
+        return QUALITY_ONE;
+
+    auto const q = (*sle)[field];
+    if (!q)
+        return QUALITY_ONE;
+    return q;
+}
+
+std::uint32_t
+DirectIOfferCrossingStep::quality (ReadView const&,
+    // set true for quality in, false for quality out
+    bool) const
+{
+    // If offer crossing then ignore trust line Quality fields.  This
+    // preserves a long-standing tradition.
+    return QUALITY_ONE;
+}
+
 std::pair<IOUAmount, bool>
-maxFlow (
-    PaymentSandbox const& sb,
-    AccountID const& src,
-    AccountID const& dst,
-    Currency const& cur)
+DirectIPaymentStep::maxFlow (ReadView const& sb, IOUAmount const&) const
+{
+    return maxPaymentFlow (sb);
+}
+
+std::pair<IOUAmount, bool>
+DirectIOfferCrossingStep::maxFlow (
+    ReadView const& sb, IOUAmount const& desired) const
+{
+    // When isLast and offer crossing then ignore trust line limits.  Offer
+    // crossing has the ability to exceed the limit set by a trust line.
+    // We presume that if someone is creating an offer then they intend to
+    // fill as much of that offer as possible, even if the offer exceeds
+    // the limit that a trust line sets.
+    //
+    // A note on using "out" as the desired parameter for maxFlow.  In some
+    // circumstances during payments we end up needing a value larger than
+    // "out" for "maxSrcToDst".  But as of now (June 2016) that never happens
+    // during offer crossing.  That's because, due to a couple of factors,
+    // "dstQIn" is always QUALITY_ONE for offer crossing.
+
+    if (isLast_)
+        return {desired, false};
+
+    return maxPaymentFlow (sb);
+}
+
+TER
+DirectIPaymentStep::check (
+    StrandContext const& ctx, std::shared_ptr<const SLE> const& sleSrc) const
+{
+    // Since this is a payment a trust line must be present.  Perform all
+    // trust line related checks.
+    {
+        auto const sleLine = ctx.view.read (keylet::line (src_, dst_, currency_));
+        if (!sleLine)
+        {
+            JLOG (j_.trace()) << "DirectStepI: No credit line. " << *this;
+            return terNO_LINE;
+        }
+
+        auto const authField = (src_ > dst_) ? lsfHighAuth : lsfLowAuth;
+
+        if (((*sleSrc)[sfFlags] & lsfRequireAuth) &&
+            !((*sleLine)[sfFlags] & authField) &&
+            (*sleLine)[sfBalance] == zero)
+        {
+            JLOG (j_.warn())
+                << "DirectStepI: can't receive IOUs from issuer without auth."
+                << " src: " << src_;
+            return terNO_AUTH;
+        }
+
+        if (ctx.prevStep &&
+            fix1449(ctx.view.info().parentCloseTime))
+        {
+            if (ctx.prevStep->bookStepBook())
+            {
+                auto const noRippleSrcToDst =
+                    ((*sleLine)[sfFlags] &
+                     ((src_ > dst_) ? lsfHighNoRipple : lsfLowNoRipple));
+                if (noRippleSrcToDst)
+                    return terNO_RIPPLE;
+            }
+        }
+    }
+
+    {
+        auto const owed = creditBalance (ctx.view, dst_, src_, currency_);
+        if (owed <= zero)
+        {
+            auto const limit = creditLimit (ctx.view, dst_, src_, currency_);
+            if (-owed >= limit)
+            {
+                JLOG (j_.debug())
+                    << "DirectStepI: dry: owed: " << owed << " limit: " << limit;
+                return tecPATH_DRY;
+            }
+        }
+    }
+    return tesSUCCESS;
+}
+
+TER
+DirectIOfferCrossingStep::check (
+    StrandContext const&, std::shared_ptr<const SLE> const&) const
+{
+    // The standard checks are all we can do because any remaining checks
+    // require the existence of a trust line.  Offer crossing does not
+    // require a pre-existing trust line.
+    return tesSUCCESS;
+}
+
+//------------------------------------------------------------------------------
+
+template <class TDerived>
+std::pair<IOUAmount, bool>
+DirectStepI<TDerived>::maxPaymentFlow (ReadView const& sb) const
 {
     auto const srcOwed = toAmount<IOUAmount> (
-        accountHolds (sb, src, cur, dst, fhIGNORE_FREEZE, beast::Journal{}));
+        accountHolds (sb, src_, currency_, dst_, fhIGNORE_FREEZE, j_));
 
     if (srcOwed.signum () > 0)
         return {srcOwed, true};
 
     // srcOwed is negative or zero
-    return {creditLimit2 (sb, dst, src, cur) + srcOwed, false};
+    return {creditLimit2 (sb, dst_, src_, currency_) + srcOwed, false};
 }
 
+template <class TDerived>
 bool
-DirectStepI::redeems (ReadView const& sb, bool fwd) const
+DirectStepI<TDerived>::redeems (ReadView const& sb, bool fwd) const
 {
-    if (!fwd)
-    {
-        auto const srcOwed = accountHolds (
-            sb, src_, currency_, dst_, fhIGNORE_FREEZE, beast::Journal{});
-        return srcOwed.signum () > 0;
-    }
-    else
-    {
-        if (!cache_)
-        {
-            assert (0);
-            return false;
-        }
+    if (fwd && cache_)
         return cache_->srcRedeems;
-    }
+
+    auto const srcOwed = accountHolds (
+        sb, src_, currency_, dst_, fhIGNORE_FREEZE, j_);
+    return srcOwed.signum () > 0;
 }
 
+template <class TDerived>
 std::pair<IOUAmount, IOUAmount>
-DirectStepI::revImp (
+DirectStepI<TDerived>::revImp (
     PaymentSandbox& sb,
     ApplyView& /*afView*/,
     boost::container::flat_set<uint256>& /*ofrsToRm*/,
@@ -252,11 +498,13 @@ DirectStepI::revImp (
 
     bool srcRedeems;
     IOUAmount maxSrcToDst;
+
     std::tie (maxSrcToDst, srcRedeems) =
-        maxFlow (sb, src_, dst_, currency_);
+        static_cast<TDerived const*>(this)->maxFlow (sb, out);
 
     std::uint32_t srcQOut, dstQIn;
     std::tie (srcQOut, dstQIn) = qualities (sb, srcRedeems, false);
+    assert (static_cast<TDerived const*>(this)->verifyDstQualityIn (dstQIn));
 
     Issue const srcToDstIss (currency_, srcRedeems ? dst_ : src_);
 
@@ -284,7 +532,6 @@ DirectStepI::revImp (
 
     if (srcToDst <= maxSrcToDst)
     {
-
         IOUAmount const in = mulRatio (
             srcToDst, srcQOut, QUALITY_ONE, /*roundUp*/ true);
         cache_.emplace (in, srcToDst, out, srcRedeems);
@@ -322,8 +569,9 @@ DirectStepI::revImp (
 // pass. But sometimes rounding differences cause the forward pass to
 // deliver more liquidity. Use the cached values from the reverse pass
 // to prevent this.
+template <class TDerived>
 void
-DirectStepI::setCacheLimiting (
+DirectStepI<TDerived>::setCacheLimiting (
     IOUAmount const& fwdIn,
     IOUAmount const& fwdSrcToDst,
     IOUAmount const& fwdOut,
@@ -362,8 +610,9 @@ DirectStepI::setCacheLimiting (
     cache_->srcRedeems = srcRedeems;
 };
 
+template <class TDerived>
 std::pair<IOUAmount, IOUAmount>
-DirectStepI::fwdImp (
+DirectStepI<TDerived>::fwdImp (
     PaymentSandbox& sb,
     ApplyView& /*afView*/,
     boost::container::flat_set<uint256>& /*ofrsToRm*/,
@@ -374,7 +623,7 @@ DirectStepI::fwdImp (
     bool srcRedeems;
     IOUAmount maxSrcToDst;
     std::tie (maxSrcToDst, srcRedeems) =
-        maxFlow (sb, src_, dst_, currency_);
+        static_cast<TDerived const*>(this)->maxFlow (sb, cache_->srcToDst);
 
     std::uint32_t srcQOut, dstQIn;
     std::tie (srcQOut, dstQIn) = qualities (sb, srcRedeems, true);
@@ -439,8 +688,9 @@ DirectStepI::fwdImp (
     return {cache_->in, cache_->out};
 }
 
+template <class TDerived>
 std::pair<bool, EitherAmount>
-DirectStepI::validFwd (
+DirectStepI<TDerived>::validFwd (
     PaymentSandbox& sb,
     ApplyView& afView,
     EitherAmount const& in)
@@ -459,7 +709,7 @@ DirectStepI::validFwd (
     bool srcRedeems;
     IOUAmount maxSrcToDst;
     std::tie (maxSrcToDst, srcRedeems) =
-            maxFlow (sb, src_, dst_, currency_);
+        static_cast<TDerived const*>(this)->maxFlow (sb, cache_->srcToDst);
 
     try
     {
@@ -495,58 +745,11 @@ DirectStepI::validFwd (
     return {true, EitherAmount (cache_->out)};
 }
 
-static
-std::uint32_t
-quality (
-    ReadView const& sb,
-    AccountID const& src,
-    AccountID const& dst,
-    Currency const& currency,
-    // set true for dst quality in, false for src quality out
-    bool qin)
-{
-    if (src == dst)
-        return QUALITY_ONE;
-
-    auto const sle = sb.read (
-        keylet::line (dst, src, currency));
-
-    if (!sle)
-        return QUALITY_ONE;
-
-    auto const& field = [&]() -> SF_U32 const&
-    {
-        if (qin)
-        {
-            // compute dst quality in
-            if (dst < src)
-                return sfLowQualityIn;
-            else
-                return sfHighQualityIn;
-        }
-        else
-        {
-            // compute src quality out
-            if (src < dst)
-                return sfLowQualityOut;
-            else
-                return sfHighQualityOut;
-        }
-    }();
-
-    if (! sle->isFieldPresent (field))
-        return QUALITY_ONE;
-
-    auto const q = (*sle)[field];
-    if (!q)
-        return QUALITY_ONE;
-    return q;
-}
-
 // Returns srcQOut, dstQIn
+template <class TDerived>
 std::pair<std::uint32_t, std::uint32_t>
-DirectStepI::qualities (
-    PaymentSandbox& sb,
+DirectStepI<TDerived>::qualities (
+    ReadView const& sb,
     bool srcRedeems,
     bool fwd) const
 {
@@ -556,7 +759,9 @@ DirectStepI::qualities (
             return {QUALITY_ONE, QUALITY_ONE};
 
         auto const prevStepQIn = prevStep_->lineQualityIn (sb);
-        auto srcQOut = quality (sb, src_, dst_, currency_, false);
+        auto srcQOut = static_cast<TDerived const*>(this)->quality (
+            sb, /* src quality out */ false);
+
         if (prevStepQIn > srcQOut)
             srcQOut = prevStepQIn;
         return {srcQOut, QUALITY_ONE};
@@ -565,24 +770,50 @@ DirectStepI::qualities (
     {
         // Charge a transfer rate when issuing and previous step redeems
         auto const prevStepRedeems = prevStep_ && prevStep_->redeems (sb, fwd);
+        assert (static_cast<TDerived const*>(this)->verifyPrevStepRedeems (
+            prevStepRedeems));
+
         std::uint32_t const srcQOut =
             prevStepRedeems ? transferRate (sb, src_).value : QUALITY_ONE;
-        auto dstQIn = quality (sb, src_, dst_, currency_, true);
+        auto dstQIn = static_cast<TDerived const*>(this)->quality (
+            sb, /* dst quality in */ true);
+
         if (isLast_ && dstQIn > QUALITY_ONE)
             dstQIn = QUALITY_ONE;
         return {srcQOut, dstQIn};
     }
 }
 
+template <class TDerived>
 std::uint32_t
-DirectStepI::lineQualityIn (ReadView const& v) const
+DirectStepI<TDerived>::lineQualityIn (ReadView const& v) const
 {
     // dst quality in
-    return quality (v, src_, dst_, currency_, true);
+    return static_cast<TDerived const*>(this)->quality (
+        v, /* dst quality in */ true);
 }
 
-TER DirectStepI::check (StrandContext const& ctx) const
+template <class TDerived>
+boost::optional<Quality>
+DirectStepI<TDerived>::qualityUpperBound(ReadView const& v, bool& redeems) const
 {
+    auto const prevRedeems = redeems;
+    redeems = this->redeems(v, true);
+    std::uint32_t const srcQOut =
+        (prevRedeems && !redeems) ? transferRate(v, src_).value : QUALITY_ONE;
+    auto dstQIn = static_cast<TDerived const*>(this)->quality (
+        v, /* dst quality in */ true);
+
+    if (isLast_ && dstQIn > QUALITY_ONE)
+        dstQIn = QUALITY_ONE;
+    Issue const iss{currency_, src_};
+    return Quality(getRate(STAmount(iss, srcQOut), STAmount(iss, dstQIn)));
+}
+
+template <class TDerived>
+TER DirectStepI<TDerived>::check (StrandContext const& ctx) const
+{
+    // The following checks apply for both payments and offer crossing.
     if (!src_ || !dst_)
     {
         JLOG (j_.debug()) << "DirectStepI: specified bad account.";
@@ -595,69 +826,35 @@ TER DirectStepI::check (StrandContext const& ctx) const
         return temBAD_PATH;
     }
 
+    auto const sleSrc = ctx.view.read (keylet::account (src_));
+    if (!sleSrc)
     {
-        auto sleSrc = ctx.view.read (keylet::account (src_));
-        if (!sleSrc)
+        JLOG (j_.warn())
+            << "DirectStepI: can't receive IOUs from non-existent issuer: "
+            << src_;
+        return terNO_ACCOUNT;
+    }
+
+    // pure issue/redeem can't be frozen
+    if (!(ctx.isLast && ctx.isFirst))
+    {
+        auto const ter = checkFreeze(ctx.view, src_, dst_, currency_);
+        if (ter != tesSUCCESS)
+            return ter;
+    }
+
+    // If previous step was a direct step then we need to check
+    // no ripple flags.
+    if (ctx.prevStep)
+    {
+        if (auto prevSrc = ctx.prevStep->directStepSrcAcct())
         {
-            JLOG (j_.warn())
-                    << "DirectStepI: can't receive IOUs from non-existent issuer: "
-                    << src_;
-            return terNO_ACCOUNT;
-        }
-
-        auto const sleLine = ctx.view.read (keylet::line (src_, dst_, currency_));
-
-        if (!sleLine)
-        {
-            JLOG (j_.trace()) << "DirectStepI: No credit line. " << *this;
-            return terNO_LINE;
-        }
-
-        auto const authField = (src_ > dst_) ? lsfHighAuth : lsfLowAuth;
-
-        if (((*sleSrc)[sfFlags] & lsfRequireAuth) &&
-            !((*sleLine)[sfFlags] & authField) &&
-            (*sleLine)[sfBalance] == zero)
-        {
-            JLOG (j_.warn())
-                << "DirectStepI: can't receive IOUs from issuer without auth."
-                << " src: " << src_;
-            return terNO_AUTH;
-        }
-
-        // pure issue/redeem can't be frozen
-        if (!(ctx.isLast && ctx.isFirst))
-        {
-            auto const ter = checkFreeze(ctx.view, src_, dst_, currency_);
+            auto const ter = checkNoRipple(
+                ctx.view, *prevSrc, src_, dst_, currency_, j_);
             if (ter != tesSUCCESS)
                 return ter;
         }
-
-        if (ctx.prevStep)
-        {
-            if (auto prevSrc = ctx.prevStep->directStepSrcAcct())
-            {
-                auto const ter = checkNoRipple(
-                    ctx.view, *prevSrc, src_, dst_, currency_, j_);
-                if (ter != tesSUCCESS)
-                    return ter;
-            }
-
-            if (fix1449(ctx.view.info().parentCloseTime))
-            {
-                if (ctx.prevStep->bookStepBook())
-                {
-                    auto const noRippleSrcToDst =
-                        ((*sleLine)[sfFlags] &
-                         ((src_ > dst_) ? lsfHighNoRipple : lsfLowNoRipple));
-                    if (noRippleSrcToDst)
-                        return terNO_RIPPLE;
-                }
-            }
-        }
     }
-
-
     {
         Issue const srcIssue{currency_, src_};
         Issue const dstIssue{currency_, dst_};
@@ -688,21 +885,7 @@ TER DirectStepI::check (StrandContext const& ctx) const
         }
     }
 
-    {
-        auto const owed = creditBalance (ctx.view, dst_, src_, currency_);
-        if (owed <= zero)
-        {
-            auto const limit = creditLimit (ctx.view, dst_, src_, currency_);
-            if (-owed >= limit)
-            {
-                JLOG (j_.debug())
-                        << "DirectStepI: dry: owed: " << owed << " limit: " << limit;
-                return tecPATH_DRY;
-            }
-        }
-    }
-
-    return tesSUCCESS;
+    return static_cast<TDerived const*>(this)->check (ctx, sleSrc);
 }
 
 //------------------------------------------------------------------------------
@@ -715,7 +898,8 @@ bool directStepEqual (Step const& step,
     AccountID const& dst,
     Currency const& currency)
 {
-    if (auto ds = dynamic_cast<DirectStepI const*> (&step))
+    if (auto ds =
+        dynamic_cast<DirectStepI<DirectIPaymentStep> const*> (&step))
     {
         return ds->src () == src && ds->dst () == dst &&
             ds->currency () == currency;
@@ -733,12 +917,25 @@ make_DirectStepI (
     AccountID const& dst,
     Currency const& c)
 {
-    // Only charge a transfer fee if the previous step redeems
-    auto r = std::make_unique<DirectStepI> (
-        src, dst, c, ctx.prevStep, ctx.isLast, ctx.j);
-    auto ter = r->check (ctx);
+    TER ter = tefINTERNAL;
+    std::unique_ptr<Step> r;
+    if (ctx.offerCrossing)
+    {
+        auto offerCrossingStep =
+            std::make_unique<DirectIOfferCrossingStep> (ctx, src, dst, c);
+        ter = offerCrossingStep->check (ctx);
+        r = std::move (offerCrossingStep);
+    }
+    else // payment
+    {
+        auto paymentStep =
+            std::make_unique<DirectIPaymentStep> (ctx, src, dst, c);
+        ter = paymentStep->check (ctx);
+        r = std::move (paymentStep);
+    }
     if (ter != tesSUCCESS)
         return {ter, nullptr};
+
     return {tesSUCCESS, std::move (r)};
 }