Improve submission order to guarantee progress and improve testbench

7 files changed, 155 insertions, 162 deletions

diff --git a/bank/src/main/kotlin/tech/libeufin/bank/db/NotificationWatcher.kt b/bank/src/main/kotlin/tech/libeufin/bank/db/NotificationWatcher.kt
index 8d6787a0..eb95e2c4 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/db/NotificationWatcher.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/db/NotificationWatcher.kt
@@ -103,7 +103,7 @@ internal class NotificationWatcher(private val pgSource: PGSimpleDataSource) {
                             }
                         }
                     } catch (e: Exception) {
-                        logger.warn("$e")
+                        e.fmtLog(logger)
                         delay(backoff.next())
                     }
                 }
diff --git a/common/src/main/kotlin/Cli.kt b/common/src/main/kotlin/Cli.kt
index fcf17688..ae9d8e87 100644
--- a/common/src/main/kotlin/Cli.kt
+++ b/common/src/main/kotlin/Cli.kt
@@ -37,7 +37,7 @@ import kotlinx.coroutines.*
 
 private val logger: Logger = LoggerFactory.getLogger("libeufin-config")
 
-fun Throwable.fmtLog(logger: Logger) {
+fun Throwable.fmt(): String{
     var msg = StringBuilder(message ?: this::class.simpleName)
     var cause = cause
     while (cause != null) {
@@ -45,7 +45,11 @@ fun Throwable.fmtLog(logger: Logger) {
         msg.append(cause.message ?: cause::class.simpleName)
         cause = cause.cause
     }
-    logger.error(msg.toString())
+    return msg.toString()
+}
+
+fun Throwable.fmtLog(logger: Logger) {
+    logger.error(this.fmt())
     logger.trace("", this)
 }
 
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
index 765037e4..47c021cd 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
@@ -132,9 +132,9 @@ suspend fun ingestIncomingPayment(
                 Instant.now()
             )
             if (result.new) {
-                logger.info("$payment bounced in '${result.bounceId}': ${e.message}")
+                logger.info("$payment bounced in '${result.bounceId}': ${e.fmt()}")
             } else {
-                logger.debug("$payment already seen and bounced in '${result.bounceId}': ${e.message}")
+                logger.debug("$payment already seen and bounced in '${result.bounceId}': ${e.fmt()}")
             }
         }
     )
@@ -154,7 +154,7 @@ private suspend fun ingestDocument(
                         is TxNotification.Incoming -> ingestIncomingPayment(db, it.payment)
                         is TxNotification.Outgoing -> ingestOutgoingPayment(db, it.payment)
                         is TxNotification.Reversal -> {
-                            logger.warn("BOUNCE '${it.msgId}': ${it.reason}")
+                            logger.error("BOUNCE '${it.msgId}': ${it.reason}")
                             db.initiated.reversal(it.msgId, "Payment bounced: ${it.reason}")
                         }
                     }
@@ -176,7 +176,7 @@ private suspend fun ingestDocument(
                     HacAction.ORDER_HAC_FINAL_NEG -> {
                         logger.debug("$ack")
                         db.initiated.logFailure(ack.orderId!!)?.let { (requestUID, msg) ->
-                            logger.warn("Payment '$requestUID' refused at ${ack.timestamp.fmtDateTime()}${if (msg != null) ": $msg" else ""}")
+                            logger.error("Payment '$requestUID' refused at ${ack.timestamp.fmtDateTime()}${if (msg != null) ": $msg" else ""}")
                         }
                     }
                     else -> {
@@ -194,7 +194,7 @@ private suspend fun ingestDocument(
             logger.debug("$status")
             if (status.paymentCode == ExternalPaymentGroupStatusCode.RJCT) {
                 db.initiated.bankFailure(status.msgId, msg)
-                logger.warn("Transaction '${status.msgId}' was rejected : $msg")
+                logger.error("Transaction '${status.msgId}' was rejected : $msg")
             } else {
                 db.initiated.bankMessage(status.msgId, msg)
             }
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
index 656ce694..8366f064 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSubmit.kt
@@ -110,7 +110,7 @@ private suspend fun submitBatch(
     ctx: SubmissionContext,
     db: Database,
 ) {
-    db.initiated.submittableGet(ctx.cfg.currency).forEach {
+    db.initiated.submittable(ctx.cfg.currency).forEach {
         logger.debug("Submitting payment '${it.requestUid}'")
         runCatching { submitInitiatedPayment(ctx, it) }.fold(
             onSuccess = { orderId -> 
@@ -119,7 +119,7 @@ private suspend fun submitBatch(
             },
             onFailure = { e ->
                 db.initiated.submissionFailure(it.id, Instant.now(), e.message)
-                logger.warn("Payment '${it.requestUid}' submission failure: ${e.message}")
+                logger.error("Payment '${it.requestUid}' submission failure: ${e.fmt()}")
                 throw e
             }
         )
@@ -175,7 +175,7 @@ class EbicsSubmit : CliktCommand("Submits any initiated payment found in the dat
                 try {
                     submitBatch(ctx, db)
                 } catch (e: Exception) {
-                    throw Exception("Failed to submit payments", e)
+                    throw Exception("Failed to submit payments")
                 }
                 // TODO take submitBatch taken time in the delay
                 delay(((frequency?.inSeconds ?: 0) * 1000).toLong())
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/db/InitiatedDAO.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/db/InitiatedDAO.kt
index efe98f82..4df7f0c9 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/db/InitiatedDAO.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/db/InitiatedDAO.kt
@@ -22,6 +22,7 @@ package tech.libeufin.nexus.db
 import tech.libeufin.nexus.*
 import tech.libeufin.common.*
 import java.time.Instant
+import java.sql.ResultSet
 
 /** Data access logic for initiated outgoing payments */
 class InitiatedDAO(private val db: Database) {
@@ -169,28 +170,15 @@ class InitiatedDAO(private val db: Database) {
         stmt.execute()
     }
 
-    // TODO WIP
-    suspend fun submittableGet(currency: String): List<InitiatedPayment> = db.conn { conn ->
-        val stmt = conn.prepareStatement("""
-            SELECT
-              initiated_outgoing_transaction_id
-             ,(amount).val as amount_val
-             ,(amount).frac as amount_frac
-             ,wire_transfer_subject
-             ,credit_payto_uri
-             ,initiation_time
-             ,request_uid
-             FROM initiated_outgoing_transactions
-             WHERE (submitted='unsubmitted' OR submitted='transient_failure')
-               AND ((amount).val != 0 OR (amount).frac != 0);
-        """)
-        stmt.all {
+    /** List every initiated payment pending submission in ther order they should be submitted */
+    suspend fun submittable(currency: String): List<InitiatedPayment> = db.conn { conn ->
+        fun extract(it: ResultSet): InitiatedPayment {
             val rowId = it.getLong("initiated_outgoing_transaction_id")
             val initiationTime = it.getLong("initiation_time").microsToJavaInstant()
             if (initiationTime == null) { // nexus fault
                 throw Exception("Found invalid timestamp at initiated payment with ID: $rowId")
             }
-            InitiatedPayment(
+            return InitiatedPayment(
                 id = it.getLong("initiated_outgoing_transaction_id"),
                 amount = it.getAmount("amount", currency),
                 creditPaytoUri = it.getString("credit_payto_uri"),
@@ -199,5 +187,31 @@ class InitiatedDAO(private val db: Database) {
                 requestUid = it.getString("request_uid")
             )
         }
+        val selectPart = """
+            SELECT
+                initiated_outgoing_transaction_id
+                ,(amount).val as amount_val
+                ,(amount).frac as amount_frac
+                ,wire_transfer_subject
+                ,credit_payto_uri
+                ,initiation_time
+                ,request_uid
+            FROM initiated_outgoing_transactions
+        """
+        // We want to maximize the number of successfully submitted transactions in the event 
+        // of a malformed transaction or a persistent error classified as transient. We send 
+        // the unsubmitted transactions first, starting with the oldest by creation time.
+        // This is the happy  path, giving every transaction a chance while being fair on the
+        // basis of creation date. 
+        // Then we retry the failed transaction, starting with the oldest by submission time.
+        // This the bad path retrying each failed transaction applying a rotation based on 
+        // resubmission time.
+        val unsubmitted = conn.prepareStatement(
+            "$selectPart WHERE submitted='unsubmitted' ORDER BY initiation_time"
+        ).all(::extract)
+        val failed = conn.prepareStatement(
+            "$selectPart WHERE submitted='transient_failure' ORDER BY last_submission_time"
+        ).all(::extract)
+        unsubmitted + failed
     }
 }
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/DatabaseTest.kt b/nexus/src/test/kotlin/DatabaseTest.kt
index c9d70ae0..6809c285 100644
--- a/nexus/src/test/kotlin/DatabaseTest.kt
+++ b/nexus/src/test/kotlin/DatabaseTest.kt
@@ -162,85 +162,60 @@ class PaymentInitiationsTest {
         db.initiated.reversal("PAY5", "status reversal")
     }
 
-    // Tests creation, unique constraint violation handling, and
-    // retrieving only one non-submitted payment.
     @Test
-    fun paymentInitiation() = setup { db, _ -> 
-        val beEmpty = db.initiated.submittableGet("KUDOS") // expect no records.
-        assertEquals(beEmpty.size, 0)
-        val initPay = InitiatedPayment(
-            id = -1,
-            amount = TalerAmount(44, 0, "KUDOS"),
-            creditPaytoUri = "payto://iban/CH9300762011623852957?receiver-name=Test",
-            wireTransferSubject = "test",
-            requestUid = "unique",
-            initiationTime = Instant.now()
-        )
-        assertEquals(db.initiated.create(initPay), PaymentInitiationResult.SUCCESS)
-        assertEquals(db.initiated.create(initPay), PaymentInitiationResult.REQUEST_UID_REUSE)
-        val haveOne = db.initiated.submittableGet("KUDOS")
-        assertTrue("Size ${haveOne.size} instead of 1") {
-            haveOne.size == 1
-                    && haveOne.first().id == 1L
-                    && haveOne.first().requestUid == "unique"
+    fun submittable() = setup { db, _ -> 
+        for (i in 0..5) {
+            assertEquals(
+                PaymentInitiationResult.SUCCESS,
+                db.initiated.create(genInitPay(requestUid = "PAY$i"))
+            )
         }
-    }
-
-    /**
-     * The SQL that gets submittable payments checks multiple
-     * statuses from them.  Checking it here.
-     */
-    @Test
-    fun submittablePayments() = setup { db, _ -> 
-        val beEmpty = db.initiated.submittableGet("KUDOS")
-        assertEquals(0, beEmpty.size)
         assertEquals(
-            db.initiated.create(genInitPay(requestUid = "first")),
-            PaymentInitiationResult.SUCCESS
+            listOf("PAY0", "PAY1", "PAY2", "PAY3", "PAY4", "PAY5"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
         )
+
+        // Check submitted not submitable
+        db.initiated.submissionSuccess(1, Instant.now(), "ORDER1")
         assertEquals(
-            db.initiated.create(genInitPay(requestUid = "second")),
-            PaymentInitiationResult.SUCCESS
+            listOf("PAY1", "PAY2", "PAY3", "PAY4", "PAY5"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
         )
+
+        // Check transient failure submitable last
+        db.initiated.submissionFailure(2, Instant.now(), "Failure")
         assertEquals(
-            db.initiated.create(genInitPay(requestUid = "third")),
-            PaymentInitiationResult.SUCCESS
+            listOf("PAY2", "PAY3", "PAY4", "PAY5", "PAY1"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
         )
 
-        // Setting the first as "transient_failure", must be found.
-        db.initiated.submissionSuccess(1, Instant.now(), "Failure")
-        // Setting the second as "success", must not be found.
-        db.initiated.submissionSuccess(2, Instant.now(), "ORDER1234")
-        val expectTwo = db.initiated.submittableGet("KUDOS")
-        // the third initiation keeps the default "unsubmitted"
-        // state, must be found.  Total 2.
-        assertEquals(1, expectTwo.size)
-    }
-
-    // Tests how the fetch method gets the list of
-    // multiple unsubmitted payment initiations.
-    @Test
-    fun paymentInitiationsMultiple() = setup { db, _ -> 
-        assertEquals(db.initiated.create(genInitPay("#1", "unique1")), PaymentInitiationResult.SUCCESS)
-        assertEquals(db.initiated.create(genInitPay("#2", "unique2")), PaymentInitiationResult.SUCCESS)
-        assertEquals(db.initiated.create(genInitPay("#3", "unique3")), PaymentInitiationResult.SUCCESS)
-        assertEquals(db.initiated.create(genInitPay("#4", "unique4")), PaymentInitiationResult.SUCCESS)
-
-        // Marking one as submitted, hence not expecting it in the results.
-        db.conn { conn ->
-            conn.execSQLUpdate("""
-                UPDATE initiated_outgoing_transactions
-                    SET submitted='success'
-                    WHERE initiated_outgoing_transaction_id=3;
-            """.trimIndent())
-        }
+        // Check persistent failure not submitable
+        db.initiated.bankFailure("PAY3", "status failure")
+        assertEquals(
+            listOf("PAY2", "PAY4", "PAY5", "PAY1"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
+        )
+        db.initiated.reversal("PAY4", "status reversal")
+        assertEquals(
+            listOf("PAY2", "PAY5", "PAY1"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
+        )
 
-        // Expecting all the payments BUT the #3 in the result.
-        db.initiated.submittableGet("KUDOS").apply {
-            assertEquals(3, this.size)
-            assertEquals("#1", this[0].wireTransferSubject)
-            assertEquals("#2", this[1].wireTransferSubject)
-            assertEquals("#4", this[2].wireTransferSubject)
-        }
+        // Check rotation
+        db.initiated.submissionFailure(3, Instant.now(), "Failure")
+        assertEquals(
+            listOf("PAY5", "PAY1", "PAY2"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
+        )
+        db.initiated.submissionFailure(6, Instant.now(), "Failure")
+        assertEquals(
+            listOf("PAY1", "PAY2", "PAY5"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
+        )
+        db.initiated.submissionFailure(2, Instant.now(), "Failure")
+        assertEquals(
+            listOf("PAY2", "PAY5", "PAY1"),
+            db.initiated.submittable("KUDOS").map { it.requestUid }
+        )
     }
 }
\ No newline at end of file
diff --git a/testbench/src/main/kotlin/Main.kt b/testbench/src/main/kotlin/Main.kt
index 1c21d8da..cc553ee8 100644
--- a/testbench/src/main/kotlin/Main.kt
+++ b/testbench/src/main/kotlin/Main.kt
@@ -47,20 +47,14 @@ fun ask(question: String): String? {
     return readlnOrNull()
 }
 
-fun CliktCommandTestResult.result() {
-    if (statusCode != 0) {
-        print("\u001b[;31mERROR:\n$output\u001b[0m")
+fun CliktCommand.run(arg: String): Boolean {
+    val res = this.test(arg)
+    if (res.statusCode != 0) {
+        println("\u001b[;31mERROR ${res.statusCode}\u001b[0m")
+    } else {
+        println("\u001b[;32mOK\u001b[0m")
     }
-}
-
-fun CliktCommandTestResult.assertOk(msg: String? = null) {
-    println("$output")
-    assertEquals(0, statusCode, msg)
-}
-
-fun CliktCommandTestResult.assertErr(msg: String? = null) {
-    println("$output")
-    assertEquals(1, statusCode, msg)
+    return res.statusCode == 0
 }
 
 data class Kind(val name: String, val settings: String?) {
@@ -121,36 +115,29 @@ class Cli : CliktCommand("Run integration tests on banks provider") {
         runBlocking {
             step("Init ${kind.name}")
 
-            nexusCmd.test("dbinit $flags").assertOk()
+            assert(nexusCmd.run("dbinit $flags"))
 
             val cmds = buildMap<String, suspend () -> Unit> {
-                put("reset-db", suspend {
-                    nexusCmd.test("dbinit -r $flags").assertOk()
-                })
-                put("recover", suspend {
-                    step("Recover old transactions")
-                    nexusCmd.test("ebics-fetch $ebicsFlags --pinned-start 2022-01-01 notification").result()
-                })
-                put("fetch", suspend {
-                    step("Fetch all documents")
-                    nexusCmd.test("ebics-fetch $ebicsFlags").result()
-                })
-                put("ack", suspend {
-                    step("Fetch CustomerAcknowledgement")
-                    nexusCmd.test("ebics-fetch $ebicsFlags acknowledgement").result()
-                })
-                put("status", suspend {
-                    step("Fetch CustomerPaymentStatusReport")
-                    nexusCmd.test("ebics-fetch $ebicsFlags status").result()
-                })
-                put("notification", suspend {
-                    step("Fetch BankToCustomerDebitCreditNotification")
-                    nexusCmd.test("ebics-fetch $ebicsFlags notification").result()
-                })
-                put("submit", suspend {
-                    step("Submit pending transactions")
-                    nexusCmd.test("ebics-submit $ebicsFlags").result()
-                })
+                fun put(name: String, args: String) {
+                    put(name, suspend {
+                        nexusCmd.run(args)
+                        Unit
+                    })
+                }
+                fun put(name: String, step: String, args: String) {
+                    put(name, suspend {
+                        step(step)
+                        nexusCmd.run(args)
+                        Unit
+                    })
+                }
+                put("reset-db", "dbinit -r $flags")
+                put("recover", "Recover old transactions", "ebics-fetch $ebicsFlags --pinned-start 2022-01-01 notification")
+                put("fetch", "Fetch all documents", "ebics-fetch $ebicsFlags")
+                put("ack", "Fetch CustomerAcknowledgement", "ebics-fetch $ebicsFlags acknowledgement")
+                put("status", "Fetch CustomerPaymentStatusReport", "ebics-fetch $ebicsFlags status")
+                put("notification", "Fetch BankToCustomerDebitCreditNotification", "ebics-fetch $ebicsFlags notification")
+                put("submit", "Submit pending transactions", "ebics-submit $ebicsFlags")
                 if (kind.test) {
                     put("reset-keys", suspend {
                         clientKeysPath.deleteIfExists()
@@ -159,46 +146,59 @@ class Cli : CliktCommand("Run integration tests on banks provider") {
                     })
                     put("tx", suspend {
                         step("Submit one transaction")
-                        nexusCmd.test("initiate-payment $flags \"$payto&amount=CHF:42&message=single%20transaction%20test\"").assertOk()
-                        nexusCmd.test("ebics-submit $ebicsFlags").assertOk()
+                        nexusCmd.run("initiate-payment $flags \"$payto&amount=CHF:42&message=single%20transaction%20test\"")
+                        nexusCmd.run("ebics-submit $ebicsFlags")
+                        Unit
                     })
                     put("txs", suspend {
                         step("Submit many transaction")
                         repeat(4) {
-                            nexusCmd.test("initiate-payment $flags --amount=CHF:${100L+it} --subject \"multi transaction test $it\" \"$payto\"").assertOk()
+                            nexusCmd.run("initiate-payment $flags --amount=CHF:${100L+it} --subject \"multi transaction test $it\" \"$payto\"")
                         }
-                        nexusCmd.test("ebics-submit $ebicsFlags").assertOk()
+                        nexusCmd.run("ebics-submit $ebicsFlags")
+                        Unit
                     })
                 } else {
                     put("tx", suspend {
                         step("Submit new transaction")
                         // TODO interactive payment editor
-                        nexusCmd.test("initiate-payment $flags \"$payto&amount=CHF:1.1&message=single%20transaction%20test\"").assertOk()
-                        nexusCmd.test("ebics-submit $ebicsFlags").assertOk()
+                        nexusCmd.run("initiate-payment $flags \"$payto&amount=CHF:1.1&message=single%20transaction%20test\"")
+                        nexusCmd.run("ebics-submit $ebicsFlags")
+                        Unit
                     })
                 }
             }
 
             while (true) {
-                var hasClientKeys = clientKeysPath.exists()
-                var hasBankKeys = bankKeysPath.exists()
-
-                if (!hasClientKeys) {
-                    if (kind.test) {
-                        step("Test INI order")
-                        ask("Got to ${kind.settings} and click on 'Reset EBICS user'.\nPress Enter when done>")
-                        nexusCmd.test("ebics-setup $flags")
-                            .assertErr("ebics-setup should failed the first time")
-                        ask("Got to ${kind.settings} and click on 'Activate EBICS user'.\nPress Enter when done>")
-                    } else {
+                // EBICS setup
+                while (true) {
+                    var clientKeys = loadClientKeys(clientKeysPath)
+                    var bankKeys = loadBankKeys(bankKeysPath)
+                    if (!kind.test && clientKeys == null) {
                         throw Exception("Clients keys are required to run netzbon tests")
+                    } else if (clientKeys == null || !clientKeys.submitted_ini) {
+                        step("Run INI and HIA order")
+                    } else if (!clientKeys.submitted_hia) {
+                        step("Run HIA order")
+                    } else if (bankKeys == null || !bankKeys.accepted) {
+                        step("Run HPB order")
+                        if (kind.test)
+                            ask("Got to ${kind.settings} and click on 'Activate EBICS user'.\nPress Enter when done>")
+                    } else {
+                        break
+                    }
+                    if (!nexusCmd.run("ebics-setup --auto-accept-keys $flags")) {
+                        clientKeys = loadClientKeys(clientKeysPath)
+                        if (kind.test) {
+                            if (clientKeys == null || !clientKeys.submitted_ini || !clientKeys.submitted_hia) {
+                                ask("Got to ${kind.settings} and click on 'Reset EBICS user'.\nPress Enter when done>")
+                            } else {
+                                ask("Got to ${kind.settings} and click on 'Activate EBICS user'.\nPress Enter when done>")
+                            }
+                        } else {
+                            ask("Activate your keys at your bank.\nPress Enter when done>")
+                        }
                     }
-                } 
-    
-                if (!hasBankKeys) {
-                    step("Test HIA order")
-                    nexusCmd.test("ebics-setup --auto-accept-keys $flags")
-                        .assertOk("ebics-setup should succeed the second time")
                 }
 
                 val arg = ask("testbench> ")!!.trim()