Merge 4.1.3 to main

.github/workflows/on_pull_request.yml

@@ -24,12 +24,9 @@ jobs:
       group: ap-test-job
       cancel-in-progress: false
 
-  codeql_analysis:
-    uses: ./.github/workflows/sub_codeql_analysis.yml
-
   complete:
     if: always()
-    needs: [ gradle_build, essential_tests, extended_tests, rust_build, codeql_analysis ]
+    needs: [ gradle_build, essential_tests, extended_tests, rust_build ]
     runs-on: ubuntu-22.04
     steps:
       - if: contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled')

.github/workflows/sub_extended_tests.yml

@@ -68,31 +68,6 @@ jobs:
       - name: Run Kafka, Postgres, and Sep24 UI with docker compose
         run: docker compose -f /home/runner/anchor-platform/service-runner/src/main/resources/docker-compose-test.yaml up -d --build
 
-      # `custody` Tests
-      - name: Start `custody` configuration
-        env:
-          TEST_PROFILE_NAME: custody
-          KT_REFERENCE_SERVER_CONFIG: /home/runner/anchor-platform/service-runner/src/main/resources/config/reference-config.yaml
-        run: |
-          cd /home/runner/anchor-platform
-          ./gradlew startServersWithTestProfile &
-          echo "PID=$!" >> $GITHUB_ENV
-
-      - name: Wait for the sep server to start and get ready
-        uses: mydea/action-wait-for-api@v1
-        with:
-          url: "http://localhost:8080/.well-known/stellar.toml"
-          interval: "1"
-          timeout: "600"
-
-      - name: Run `custody` configuration tests
-        env:
-          TEST_PROFILE_NAME: custody
-        run: |
-          cd /home/runner/anchor-platform
-          ./gradlew :service-runner:clean :extended-tests:clean :extended-tests:test --tests org.stellar.anchor.platform.suite.CustodyTestSuite         
-          kill -9 $PID
-
       # `rpc` Tests
       - name: Start `rpc` configuration
         env:

README.md

@@ -1,6 +1,6 @@
 [![License](https://badgen.net/badge/license/Apache%202/blue?icon=github&label=License)](https://github.com/stellar/anchor-platform/blob/develop/LICENSE)
 [![GitHub Version](https://badgen.net/github/release/stellar/anchor-platform?icon=github&label=Latest%20release)](https://github.com/stellar/anchor-platform/releases)
-[![Docker](https://badgen.net/badge/Latest%20Release/v4.1.1/blue?icon=docker)](https://hub.docker.com/r/stellar/anchor-platform/tags?page=1&name=4.1.1)
+[![Docker](https://badgen.net/badge/Latest%20Release/v4.1.3/blue?icon=docker)](https://hub.docker.com/r/stellar/anchor-platform/tags?page=1&name=4.1.3)
 ![Develop Branch](https://github.com/stellar/anchor-platform/actions/workflows/on_push_to_develop.yml/badge.svg?branch=develop)
 
 <div style="text-align: center">

build.gradle.kts

@@ -213,7 +213,7 @@ subprojects {
 
 allprojects {
   group = "org.stellar.anchor-sdk"
-  version = "4.1.1"
+  version = "4.1.3"
 
   tasks.jar {
     manifest {

kotlin-reference-server/src/main/kotlin/org/stellar/reference/sep24/DepositService.kt

@@ -4,13 +4,13 @@ import io.github.oshai.kotlinlogging.KotlinLogging
 import java.math.BigDecimal
 import java.math.RoundingMode
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.retryWhen
+import kotlinx.coroutines.withTimeout
 import org.stellar.reference.client.PaymentClient
 import org.stellar.reference.data.*
 import org.stellar.reference.service.SepHelper
 import org.stellar.reference.transactionWithRetry
 import org.stellar.sdk.Asset
+import org.stellar.sdk.responses.sorobanrpc.GetTransactionResponse
 
 private val log = KotlinLogging.logger {}
 
@@ -23,7 +23,7 @@ class DepositService(private val cfg: Config, private val paymentClient: Payment
     amount: BigDecimal,
     account: String,
     asset: String,
-    memo: String?
+    memo: String?,
   ) {
     try {
       var transaction = sep24.getTransaction(transactionId)
@@ -51,18 +51,19 @@ class DepositService(private val cfg: Config, private val paymentClient: Payment
         // 7. Finalize custody Stellar anchor transaction
         finalizeCustodyStellarTransaction(transactionId)
       } else {
+        lateinit var txHash: String
         transactionWithRetry {
-          val txHash =
+          txHash =
             paymentClient.send(
               account,
               Asset.create(asset.replace("stellar:", "")),
               transaction.amountOut!!.amount!!,
-              memo
+              memo,
             )
-
-          // 6. Finalize Stellar anchor transaction
-          finalizeStellarTransaction(transactionId, txHash)
         }
+        waitForValidTransaction(paymentClient, txHash)
+        // 6. Finalize Stellar anchor transaction
+        finalizeStellarTransaction(transaction, txHash)
       }
 
       log.info { "Transaction completed: $transactionId" }
@@ -78,6 +79,37 @@ class DepositService(private val cfg: Config, private val paymentClient: Payment
     }
   }
 
+  private suspend fun waitForValidTransaction(
+    paymentClient: PaymentClient,
+    txHash: String,
+    maxAttempts: Int = 10,
+    timeoutMs: Long? = null,
+  ): GetTransactionResponse {
+    suspend fun poll(): GetTransactionResponse {
+      var attempt = 0
+      while (attempt < maxAttempts) {
+        val txn = paymentClient.getTransaction(txHash)
+        if (txn?.status == GetTransactionResponse.GetTransactionStatus.SUCCESS) {
+          return txn
+        }
+        if (txn?.status == GetTransactionResponse.GetTransactionStatus.FAILED) {
+          throw IllegalStateException("Transaction $txHash failed")
+        }
+        attempt++
+        delay(1000L)
+      }
+      throw IllegalStateException(
+        "Transaction $txHash did not reach a valid status after $maxAttempts attempts"
+      )
+    }
+
+    return if (timeoutMs != null) {
+      withTimeout(timeoutMs) { poll() }
+    } else {
+      poll()
+    }
+  }
+
   private suspend fun initiateTransfer(transactionId: String, amount: BigDecimal, asset: String) {
     val fee = calculateFee(amount)
     val stellarAsset = "stellar:$asset"
@@ -123,11 +155,6 @@ class DepositService(private val cfg: Config, private val paymentClient: Payment
         "pending_anchor",
         "funds received, transaction is being processed",
       )
-      sep24.patchTransaction(
-        transactionId,
-        "pending_stellar",
-        "funds received, transaction is being processed",
-      )
     }
   }
 
@@ -148,30 +175,25 @@ class DepositService(private val cfg: Config, private val paymentClient: Payment
   }
 
   private suspend fun finalizeStellarTransaction(
-    transactionId: String,
-    stellarTransactionId: String
+    transaction: Transaction,
+    stellarTransactionId: String,
   ) {
     // SAC transfers submitted to RPC are asynchronous, we will need to retry
     // until the RPC returns a success response
     if (cfg.appSettings.rpcEnabled) {
-      flow<Unit> {
-          sep24.rpcAction(
-            "notify_onchain_funds_sent",
-            NotifyOnchainFundsSentRequest(
-              transactionId = transactionId,
-              stellarTransactionId = stellarTransactionId,
-            ),
-          )
-        }
-        .retryWhen { _, attempt ->
-          if (attempt < 5) {
-            delay(5_000)
-            return@retryWhen true
-          } else {
-            return@retryWhen false
-          }
+      if (transaction.status == "pending_anchor") {
+        sep24.rpcAction(
+          "notify_onchain_funds_sent",
+          NotifyOnchainFundsSentRequest(
+            transactionId = transaction.id,
+            stellarTransactionId = stellarTransactionId,
+          ),
+        )
+      } else {
+        log.warn {
+          "Transaction ${transaction.id} is in unexpected status ${transaction.status}, skipping notify_onchain_funds_sent"
         }
-        .collect {}
+      }
     }
   }
 

platform/src/main/java/org/stellar/anchor/platform/observer/stellar/StellarRpcPaymentObserver.java

@@ -17,7 +17,6 @@
 import java.util.stream.Stream;
 import lombok.Builder;
 import lombok.Getter;
-import lombok.val;
 import org.stellar.anchor.api.asset.AssetInfo;
 import org.stellar.anchor.api.asset.StellarAssetInfo;
 import org.stellar.anchor.api.exception.AnchorException;
@@ -119,12 +118,14 @@ private void fetchEvents() {
 
     try {
       GetEventsResponse response = sorobanServer.getEvents(buildEventRequest(cursor));
+      metricLatestBlockRead.set(response.getLatestLedger());
       if (response.getEvents() != null && !response.getEvents().isEmpty()) {
         processEvents(response.getEvents());
       }
       // Save the cursor for the next request
       cursor = response.getCursor();
       saveCursor(cursor);
+      metricLatestBlockProcessed.set(response.getLatestLedger());
     } catch (IOException ioex) {
       warnF(
           "Error fetching latest ledger: {}. ex={}. Wait for next retry.",
@@ -136,17 +137,13 @@ private void fetchEvents() {
   private void processEvents(List<EventInfo> events) {
     if (events == null || events.isEmpty()) return;
     debugF("Processing {} 'transfer' events", events.size());
-    val lastEvent = events.get(events.size() - 1);
-    if (lastEvent != null) metricLatestBlockRead.set(lastEvent.getLedger());
 
     for (EventInfo event : events) {
       ShouldProcessResult result = shouldProcess(event);
       if (result.shouldProcess) {
         processTransferEvent(result);
       }
     }
-
-    if (lastEvent != null) metricLatestBlockProcessed.set(lastEvent.getLedger());
   }
 
   private void processTransferEvent(ShouldProcessResult result) {

service-runner/src/main/resources/version-info.properties

@@ -1 +1 @@
-version=4.1.1
+version=4.1.3