Repository: hyperledger/quilt
Pull request #398 (2019-12-06 01:52:37)

Is/370 only fail on 4xx

codecov[bot]: # [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=h1) Report
> Merging [#398](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=desc) into [master](https://codecov.io/gh/hyperledger/quilt/commit/861887dcbdf1a4d10259ba73f334aa8dc0c4fa09?src=pr&el=desc) will **decrease** coverage by `0.12%`.
> The diff coverage is `87.5%`.

[![Impacted file tree graph](https://codecov.io/gh/hyperledger/quilt/pull/398/graphs/tree.svg?width=650&token=XEfCq6zC6W&height=150&src=pr)](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##             master    #398      +/-   ##
===========================================
- Coverage     82.52%   82.4%   -0.13%     
  Complexity      829     829              
===========================================
  Files           249     249              
  Lines          3754    3762       +8     
  Branches        229     231       +2     
===========================================
+ Hits           3098    3100       +2     
- Misses          560     568       +8     
+ Partials         96      94       -2
```


| [Impacted Files](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...ava/org/interledger/link/http/IlpOverHttpLink.java](https://codecov.io/gh/hyperledger/quilt/pull/398/diff?src=pr&el=tree#diff-bGluay1wYXJlbnQvbGluay1pbHAtb3Zlci1odHRwL3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL2xpbmsvaHR0cC9JbHBPdmVySHR0cExpbmsuamF2YQ==) | `87.61% <66.66%> (-0.12%)` | `14 <0> (ø)` | |
| [.../interledger/stream/sender/SimpleStreamSender.java](https://codecov.io/gh/hyperledger/quilt/pull/398/diff?src=pr&el=tree#diff-c3RyZWFtLXBhcmVudC9zdHJlYW0tY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2ludGVybGVkZ2VyL3N0cmVhbS9zZW5kZXIvU2ltcGxlU3RyZWFtU2VuZGVyLmphdmE=) | `91.17% <92.3%> (-1.98%)` | `0 <0> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=footer). Last update [861887d...319382e](https://codecov.io/gh/hyperledger/quilt/pull/398?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java
index 72f9a9f..d0bd5f5 100644
--- a/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java
+++ b/link-parent/link-ilp-over-http/src/main/java/org/interledger/link/http/IlpOverHttpLink.java
@@ -42,6 +42,7 @@ import java.io.IOException;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 /**
  * <p>An extension of {@link AbstractLink} that handles HTTP (aka, ILP over HTTP) connections, both incoming and
@@ -132,17 +133,28 @@ public class IlpOverHttpLink extends AbstractLink<IlpOverHttpLinkSettings> imple
         Optional<ThrowableProblem> problem = parseThrowableProblem(preparePacket, errorResponseBody);
         final InterledgerRejectPacket rejectPacket;
 
-        if (response.code() == 401 || response.code() == 403) {
-          // If this code is returned, we know the Link is misconfigured, so throw a LinkException Exception.
-          throw new LinkException(String.format(
-              "Unable to connect to remote ILP-over-HTTP Link: Invalid Bearer Token. response=%s", response
-          ), this.getLinkId());
-        } else if (response.code() >= 400 && response.code() < 500) {
+        if (response.code() >= 400 && response.code() < 500) {
+          String customErrorMessage = null;
+          if (response.code() == 401 || response.code() == 403) {
+            // If this code is returned, we know the Link is misconfigured
+            customErrorMessage = String.format("Unable to connect to remote ILP-over-HTTP Link: Invalid Bearer " +
+              "Token. response=%s", response);
+            logger.error(customErrorMessage);
+          }
+
+          String message = Stream.of(
+            customErrorMessage,
+            problem.map(Problem::getTitle).orElse(null),
+            errorResponseBody
+          )
+          .filter(Objects::nonNull)
+          .findFirst()
+          .get();
           // The request was bad for some reason, likely due to whatever is in the packet.
           rejectPacket = InterledgerRejectPacket.builder()
               .triggeredBy(getOperatorAddressSupplier().get())
               .code(InterledgerErrorCode.F00_BAD_REQUEST)
-              .message(problem.map(Problem::getTitle).orElse(errorResponseBody))
+              .message(message)
               .build();
         } else {
           // Something else went wrong on the server...try again later.
diff --git a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
index 465ae0f..291e591 100644
--- a/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
+++ b/stream-parent/stream-client/src/main/java/org/interledger/stream/sender/SimpleStreamSender.java
@@ -1,5 +1,6 @@
 package org.interledger.stream.sender;
 
+import static org.interledger.core.InterledgerErrorCode.F00_BAD_REQUEST;
 import static org.interledger.core.InterledgerErrorCode.F08_AMOUNT_TOO_LARGE_CODE;
 import static org.interledger.core.InterledgerErrorCode.T04_INSUFFICIENT_LIQUIDITY_CODE;
 import static org.interledger.stream.StreamUtils.generatedFulfillableFulfillment;
@@ -62,7 +63,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
-
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -249,6 +249,8 @@ public class SimpleStreamSender implements StreamSender {
 
     private Optional<Denomination> receiverDenomination;
 
+    private final AtomicBoolean unrecoverableErrorEncountered;
+
     /**
      * Required-args Constructor.
      *
@@ -294,6 +296,8 @@ public class SimpleStreamSender implements StreamSender {
       this.paymentTracker = request.paymentTracker();
 
       this.receiverDenomination = Optional.empty();
+
+      this.unrecoverableErrorEncountered = new AtomicBoolean(false);
     }
 
     /**
@@ -408,7 +412,7 @@ public class SimpleStreamSender implements StreamSender {
           .data(streamPacketData)
           .build();
 
-      InterledgerResponsePacket responsePacket = link.sendPacket(preparePacket);
+      InterledgerResponsePacket responsePacket = sendPacketAndCheckForFailure(preparePacket);
 
       final Function<InterledgerResponsePacket, Optional<Denomination>> readDetails = (p) -> {
         final StreamPacket packet = this.fromEncrypted(sharedSecret, p.getData());
@@ -422,6 +426,23 @@ public class SimpleStreamSender implements StreamSender {
       return responsePacket.map(readDetails::apply, readDetails::apply);
     }
 
+    /**
+     * Send the packet but check to see if an error in the HTTP 4XX range was encountered so that we know if we
+     * should stop retrying
+     * @param preparePacket
+     * @return the returned response packet
+     */
+    @VisibleForTesting
+    protected InterledgerResponsePacket sendPacketAndCheckForFailure(InterledgerPreparePacket preparePacket) {
+      InterledgerResponsePacket response = link.sendPacket(preparePacket);
+      response.handle((fulfill) -> {}, (reject) -> {
+        if (reject.getCode().equals(F00_BAD_REQUEST)) {
+          unrecoverableErrorEncountered.set(true);
+        }
+      });
+      return response;
+    }
+
     private void sendMoneyPacketized() {
 
       final AtomicBoolean timeoutReached = new AtomicBoolean(false);
@@ -446,7 +467,7 @@ public class SimpleStreamSender implements StreamSender {
         UnsignedLong amountToSend = amounts.getAmountToSend();
         UnsignedLong receiverMinimum = amounts.getMinimumAmountToAccept();
 
-        if (amountToSend.equals(UnsignedLong.ZERO) || timeoutReached.get()) {
+        if (amountToSend.equals(UnsignedLong.ZERO) || timeoutReached.get() || unrecoverableErrorEncountered.get()) {
           try {
             // Don't send any more, but wait a bit for outstanding requests to complete so we don't cycle needlessly in
             // a while loop that doesn't do anything useful.
@@ -547,7 +568,7 @@ public class SimpleStreamSender implements StreamSender {
         executorService.submit(() -> {
           if (!timeoutReached.get()) {
             try {
-              InterledgerResponsePacket responsePacket = link.sendPacket(preparePacket);
+              InterledgerResponsePacket responsePacket = sendPacketAndCheckForFailure(preparePacket);
               responsePacket.handle(
                   fulfillPacket -> handleFulfill(preparePacket, streamPacket, fulfillPacket, prepareAmounts),
                   rejectPacket -> handleReject(preparePacket, streamPacket, rejectPacket, prepareAmounts,
@@ -586,8 +607,12 @@ public class SimpleStreamSender implements StreamSender {
       //   the connection is not closed
       //   and you haven't delivered the full amount
       //   and you haven't timed out
+      //   and you're not trying to send to much
+      //   and we haven't hit an unrecoverable error
       return this.congestionController.hasInFlight()
-          || (!streamConnection.isClosed() && paymentTracker.moreToSend() && !timeoutReached && !tryingToSendTooMuch);
+          || (
+            !streamConnection.isClosed() && paymentTracker.moreToSend() && !timeoutReached && !tryingToSendTooMuch &&
+              !unrecoverableErrorEncountered.get());
     }
 
     /**
@@ -742,6 +767,11 @@ public class SimpleStreamSender implements StreamSender {
       }
     }
 
+    @VisibleForTesting
+    protected boolean isUnrecoverableErrorEncountered() {
+      return this.unrecoverableErrorEncountered.get();
+    }
+
     ///**
     // * Close the current STREAM connection by sending a {@link ConnectionCloseFrame} to the receiver.
     // *


Test patch:
diff --git a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverHttpLinkTest.java b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverHttpLinkTest.java
index 91c626b..2a77c42 100644
--- a/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverHttpLinkTest.java
+++ b/link-parent/link-ilp-over-http/src/test/java/org/interledger/link/http/IlpOverHttpLinkTest.java
@@ -92,23 +92,24 @@ public class IlpOverHttpLinkTest {
   @Test
   public void rejectOnNotAuthed() throws Exception {
     mockCall(401);
-    expectedException.expect(LinkException.class);
-    expectedException.expectMessage("Unable to connect to remote ILP-over-HTTP Link: "
-        + "Invalid Bearer Token. response=Response{"
-        + "protocol=h2, code=401, message=stop asking me to set stuff, url=https://existentialcrisis.com/"
-        + "}");
-    link.sendPacket(packet);
+    InterledgerResponsePacket responsePacket = link.sendPacket(packet);
+    assertThat(responsePacket).extracting("code", "message")
+      .containsExactly(
+        InterledgerErrorCode.F00_BAD_REQUEST,
+        "Unable to connect to remote ILP-over-HTTP Link: Invalid Bearer Token. response=Response{protocol=h2, " +
+          "code=401, message=stop asking me to set stuff, url=https://existentialcrisis.com/}"
+      );
   }
 
   @Test
   public void rejectOnForbidden() throws Exception {
     mockCall(403);
-    expectedException.expect(LinkException.class);
-    expectedException.expectMessage("Unable to connect to remote ILP-over-HTTP Link: "
-        + "Invalid Bearer Token. response=Response{"
-        + "protocol=h2, code=403, message=stop asking me to set stuff, url=https://existentialcrisis.com/"
-        + "}");
-    link.sendPacket(packet);
+    InterledgerResponsePacket responsePacket = link.sendPacket(packet);
+    assertThat(responsePacket).extracting("code", "message")
+      .containsExactly(InterledgerErrorCode.F00_BAD_REQUEST,
+        "Unable to connect to remote ILP-over-HTTP Link: Invalid Bearer Token. response=Response{protocol=h2, " +
+          "code=403, message=stop asking me to set stuff, url=https://existentialcrisis.com/}"
+      );
   }
 
   @Test
diff --git a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
index 8edd3d2..98158b9 100644
--- a/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
+++ b/stream-parent/stream-client/src/test/java/org/interledger/stream/sender/SendMoneyAggregatorTest.java
@@ -2,6 +2,7 @@ package org.interledger.stream.sender;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atMost;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -17,6 +18,7 @@ import org.interledger.core.InterledgerFulfillment;
 import org.interledger.core.InterledgerPacketType;
 import org.interledger.core.InterledgerPreparePacket;
 import org.interledger.core.InterledgerRejectPacket;
+import org.interledger.core.InterledgerResponsePacket;
 import org.interledger.core.SharedSecret;
 import org.interledger.encoding.asn.framework.CodecContext;
 import org.interledger.link.Link;
@@ -25,6 +27,7 @@ import org.interledger.stream.Denominations;
 import org.interledger.stream.PaymentTracker;
 import org.interledger.stream.PrepareAmounts;
 import org.interledger.stream.SendMoneyRequest;
+import org.interledger.stream.SendMoneyResult;
 import org.interledger.stream.SenderAmountMode;
 import org.interledger.stream.StreamConnection;
 import org.interledger.stream.StreamConnectionClosedException;
@@ -65,7 +68,7 @@ public class SendMoneyAggregatorTest {
 
   // 5 seconds max per method tested
   @Rule
-  public Timeout globalTimeout = Timeout.seconds(5);
+  public Timeout globalTimeout = Timeout.seconds(600);
 
   @Rule
   public ExpectedException expectedException = ExpectedException.none();
@@ -279,6 +282,7 @@ public class SendMoneyAggregatorTest {
     //   and you haven't delivered the full amount
     //   and you haven't timed out
     //   and you haven't tried to send too much
+    //   and we haven't hit an unrecoverable error
 
     setSoldierOnBooleans(false, false, false);
     allSoldierOnsFalse();
@@ -306,6 +310,14 @@ public class SendMoneyAggregatorTest {
 
     setSoldierOnBooleans(true, true, true);
     allSoldierOnsTrue();
+
+    // flip flag on unrecoverable error
+    when(linkMock.sendPacket(any())).thenReturn(sampleRejectPacket(InterledgerErrorCode.F00_BAD_REQUEST));
+    sendMoneyAggregator.sendPacketAndCheckForFailure(samplePreparePacket());
+    setSoldierOnBooleans(false, false, false);
+    allSoldierOnsFalse();
+    setSoldierOnBooleans(false, false, true);
+    allSoldierOnsFalse();
   }
 
   private void allSoldierOnsTrue() {
@@ -437,6 +449,98 @@ public class SendMoneyAggregatorTest {
     verify(congestionControllerMock, times(1)).reject(UnsignedLong.ONE, reject);
   }
 
+  @Test
+  public void sendPacketAndCheckForFailureMarksUnrecoverableForF00() {
+    when(linkMock.sendPacket(any())).thenReturn(sampleRejectPacket(InterledgerErrorCode.F00_BAD_REQUEST));
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isFalse();
+    sendMoneyAggregator.sendPacketAndCheckForFailure(samplePreparePacket());
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isTrue();
+  }
+
+  @Test
+  public void preflightCheckFlagsAsUnrecoverable() throws Exception {
+    when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
+    when(linkMock.sendPacket(any())).thenReturn(sampleRejectPacket(InterledgerErrorCode.F00_BAD_REQUEST));
+    StreamPacket streamPacket = StreamPacket.builder().from(sampleStreamPacket())
+      .addFrames(StreamMoneyFrame.builder()
+        .shares(UnsignedLong.ONE)
+        .streamId(UnsignedLong.ONE)
+        .build())
+      .build();
+    when(streamCodecContextMock.read(any(), any())).thenReturn(streamPacket);
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isFalse();
+    sendMoneyAggregator.preflightCheck();
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isTrue();
+  }
+
+  @Test
+  public void stopSendingWhenUnrecoverableErrorEncountered() throws Exception {
+    SendMoneyRequest request = SendMoneyRequest.builder()
+      .sharedSecret(sharedSecret)
+      .sourceAddress(sourceAddress)
+      .senderAmountMode(SenderAmountMode.SENDER_AMOUNT)
+      .destinationAddress(destinationAddress)
+      .amount(originalAmountToSend)
+      .timeout(Optional.of(Duration.ofSeconds(60)))
+      .denomination(Denominations.XRP)
+      .paymentTracker(new FixedSenderAmountPaymentTracker(UnsignedLong.valueOf(10l), new NoOpExchangeRateCalculator()))
+      .build();
+
+    ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
+
+    this.sendMoneyAggregator = new SendMoneyAggregator(
+      executor, streamConnectionMock, streamCodecContextMock, linkMock, congestionControllerMock,
+      streamEncryptionServiceMock, request);
+
+    when(congestionControllerMock.hasInFlight()).thenAnswer(new Answer<Boolean>() {
+
+      private AtomicInteger requests = new AtomicInteger(0);
+
+      @Override
+      public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
+        return requests.incrementAndGet() <= 10;
+      }
+
+    });
+
+    when(congestionControllerMock.getMaxAmount()).thenReturn(UnsignedLong.ONE);
+
+    when(streamConnectionMock.nextSequence()).thenReturn(UnsignedLong.ONE);
+    when(linkMock.sendPacket(any())).thenAnswer(new Answer<InterledgerResponsePacket>() {
+      private AtomicInteger invocations = new AtomicInteger(0);
+
+      @Override
+      public InterledgerResponsePacket answer(InvocationOnMock invocationOnMock) throws Throwable {
+        if (invocations.incrementAndGet() <= 2) {
+          return sampleFulfillPacket();
+        }
+        return sampleRejectPacket(InterledgerErrorCode.F00_BAD_REQUEST);
+      }
+    });
+    StreamPacket streamPacket = StreamPacket.builder()
+      .prepareAmount(UnsignedLong.ONE)
+      .sequence(UnsignedLong.ZERO)
+      .interledgerPacketType(InterledgerPacketType.FULFILL)
+      .addFrames(StreamMoneyFrame.builder()
+        .shares(UnsignedLong.ONE)
+        .streamId(UnsignedLong.ONE)
+        .build())
+      .build();
+    when(streamCodecContextMock.read(any(), any())).thenReturn(streamPacket);
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isFalse();
+    SendMoneyResult result = sendMoneyAggregator.send().get();
+    assertThat(sendMoneyAggregator.isUnrecoverableErrorEncountered()).isTrue();
+    assertThat(result)
+      .extracting("amountDelivered", "amountSent", "numFulfilledPackets",
+        "successfulPayment")
+      .containsExactly(UnsignedLong.ONE, UnsignedLong.ONE, 1, false);
+    // depending on execution order, we may halt sooner than trying all 10 potential times
+    assertThat(result.numRejectPackets()).isLessThanOrEqualTo(9);
+    assertThat(result.amountLeftToSend()).isLessThanOrEqualTo(UnsignedLong.valueOf(9));
+    verify(congestionControllerMock, atMost(11)).getMaxAmount();
+    verify(linkMock, atMost(11)).sendPacket(any());
+  }
+
   /**
    * Helper method to set the soldierOn mock values for clearer test coverage.
    */