findByIdempotencyKey(String idempotencyKey);
+
+ /**
+ * Claims up to {@code pageable}'s page size of due rows for the given
+ * statuses via {@code SELECT ... FOR UPDATE}, so two schedulers (e.g. two
+ * app instances polling the same table) can never claim the same row: the
+ * second transaction's {@code SELECT} simply blocks until the first
+ * commits or rolls back, then re-evaluates the {@code WHERE} clause and
+ * (having missed the now-claimed row) returns nothing for it — no
+ * exception, no explicit retry needed by the caller.
+ *
+ * This can't deadlock against {@link EscrowOrchestrationInserter#insert},
+ * which only ever inserts new rows and never locks an existing one. It
+ * also can't deadlock against itself: every caller locks rows in the same
+ * {@code order by r.id}, so two concurrent callers each waiting on a
+ * single-row page (as {@link com.guildworkman.api.escrow.service.EscrowOrchestrationService}
+ * uses it) can never form a wait-cycle. A caller that widens the page
+ * size and processes rows out of id order would reintroduce that risk.
+ *
+ *
A stuck row (its transaction holding the lock indefinitely — e.g. a
+ * hung Soroban RPC call inside the same transaction) would simply be
+ * skipped by other callers until Postgres's own
+ * {@code lock_timeout}/{@code statement_timeout} intervenes; the
+ * scheduled callers here don't set an explicit lock-wait timeout and rely
+ * on {@link com.guildworkman.api.escrow.rpc.SorobanRpcProperties#getRequestTimeout()}
+ * to bound how long that transaction can stay open in the first place.
+ */
+ @Lock(LockModeType.PESSIMISTIC_WRITE)
+ @Query("select r from EscrowOrchestrationRequest r where r.status in :statuses and r.nextAttemptAt <= :now order by r.id")
+ List claimNext(@Param("statuses") Set statuses,
+ @Param("now") Instant now, Pageable pageable);
+
+ List findByStatusAndReconciliationStatus(
+ OrchestrationStatus status, ReconciliationStatus reconciliationStatus, Pageable pageable);
+
+ long countByStatus(OrchestrationStatus status);
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/GetTransactionResult.java b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/GetTransactionResult.java
new file mode 100644
index 0000000..ee31273
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/GetTransactionResult.java
@@ -0,0 +1,23 @@
+package com.guildworkman.api.escrow.rpc;
+
+/**
+ * Result of a Soroban RPC {@code getTransaction} call.
+ *
+ * @param status one of {@code SUCCESS}, {@code NOT_FOUND}, {@code FAILED}
+ * @param resultXdr base64 {@code TransactionResult} XDR when the transaction was found, else {@code null}
+ * @param ledger ledger sequence the transaction was included in, or {@code null} if not found
+ */
+public record GetTransactionResult(String status, String resultXdr, Long ledger) {
+
+ public boolean isSuccess() {
+ return "SUCCESS".equals(status);
+ }
+
+ public boolean isFailed() {
+ return "FAILED".equals(status);
+ }
+
+ public boolean isNotFound() {
+ return "NOT_FOUND".equals(status);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SendTransactionResult.java b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SendTransactionResult.java
new file mode 100644
index 0000000..af0a2a6
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SendTransactionResult.java
@@ -0,0 +1,19 @@
+package com.guildworkman.api.escrow.rpc;
+
+/**
+ * Result of a Soroban RPC {@code sendTransaction} call.
+ *
+ * @param hash the transaction hash (present for PENDING/DUPLICATE)
+ * @param status one of {@code PENDING}, {@code DUPLICATE}, {@code TRY_AGAIN_LATER}, {@code ERROR}
+ * @param errorResultXdr base64 {@code TransactionResult} XDR when status is {@code ERROR}, else {@code null}
+ */
+public record SendTransactionResult(String hash, String status, String errorResultXdr) {
+
+ public boolean isAccepted() {
+ return "PENDING".equals(status) || "DUPLICATE".equals(status);
+ }
+
+ public boolean isRetryable() {
+ return "TRY_AGAIN_LATER".equals(status);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcClient.java b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcClient.java
new file mode 100644
index 0000000..1b6ef31
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcClient.java
@@ -0,0 +1,144 @@
+package com.guildworkman.api.escrow.rpc;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.UUID;
+import java.util.function.Consumer;
+
+/**
+ * Thin JSON-RPC 2.0 client for the Soroban RPC methods this service needs:
+ * submitting a signed transaction and polling its outcome. Transaction
+ * envelopes and results are treated as opaque base64 XDR strings — this
+ * client never encodes or decodes XDR itself, it only relays what the
+ * caller (which builds and signs transactions client-side) hands it.
+ *
+ * Every call gets its own JSON-RPC {@code id} (a random UUID), reused as
+ * a correlation id in logs and exception messages so a single round-trip can
+ * be traced end to end. Response/error bodies are truncated before they're
+ * logged or embedded in an exception message — a Soroban error payload can
+ * echo back the (opaque, but potentially large) request XDR, and this is the
+ * only place in the request path where that payload is turned into a plain
+ * string that might land in application logs.
+ *
+ * @see Soroban RPC methods
+ */
+@Component
+public class SorobanRpcClient {
+
+ private static final Logger log = LoggerFactory.getLogger(SorobanRpcClient.class);
+ private static final MediaType JSON = MediaType.get("application/json; charset=utf-8");
+
+ /** Caps how much of a response/error body is ever logged or placed in an exception message. */
+ private static final int MAX_LOGGED_BODY_LENGTH = 500;
+
+ private final OkHttpClient httpClient;
+ private final ObjectMapper objectMapper;
+ private final SorobanRpcProperties properties;
+
+ public SorobanRpcClient(OkHttpClient httpClient, ObjectMapper objectMapper, SorobanRpcProperties properties) {
+ this.objectMapper = objectMapper;
+ this.properties = properties;
+ // properties.requestTimeout governs this Soroban-specific client only;
+ // the injected OkHttpClient bean is shared app-wide (see AppConfig) and
+ // is left with its own defaults for other callers (e.g. PaymentServiceImpl).
+ Duration timeout = properties.getRequestTimeout();
+ this.httpClient = httpClient.newBuilder()
+ .callTimeout(timeout)
+ .connectTimeout(timeout)
+ .readTimeout(timeout)
+ .writeTimeout(timeout)
+ .build();
+ }
+
+ public SendTransactionResult sendTransaction(String signedTransactionXdr) {
+ JsonNode result = call("sendTransaction", params -> params.put("transaction", signedTransactionXdr));
+ return new SendTransactionResult(
+ textOrNull(result, "hash"),
+ textOrNull(result, "status"),
+ textOrNull(result, "errorResultXdr"));
+ }
+
+ public GetTransactionResult getTransaction(String hash) {
+ JsonNode result = call("getTransaction", params -> params.put("hash", hash));
+ Long ledger = result.hasNonNull("ledger") ? result.get("ledger").asLong() : null;
+ return new GetTransactionResult(
+ textOrNull(result, "status"),
+ textOrNull(result, "resultXdr"),
+ ledger);
+ }
+
+ private JsonNode call(String method, Consumer paramsBuilder) {
+ String requestId = UUID.randomUUID().toString();
+
+ ObjectNode params = objectMapper.createObjectNode();
+ paramsBuilder.accept(params);
+
+ ObjectNode body = objectMapper.createObjectNode();
+ body.put("jsonrpc", "2.0");
+ body.put("id", requestId);
+ body.put("method", method);
+ body.set("params", params);
+
+ Request request = new Request.Builder()
+ .url(properties.getUrl())
+ .post(RequestBody.create(body.toString(), JSON))
+ .build();
+
+ log.debug("Soroban RPC request rpcId={} method={}", requestId, method);
+
+ try (Response response = httpClient.newCall(request).execute()) {
+ if (response.body() == null) {
+ throw new SorobanRpcException("Soroban RPC rpcId=" + requestId + " method=" + method
+ + " returned an empty response");
+ }
+ String responseBody = response.body().string();
+ if (!response.isSuccessful()) {
+ throw new SorobanRpcException("Soroban RPC rpcId=" + requestId + " method=" + method
+ + " HTTP " + response.code() + ": " + truncate(responseBody));
+ }
+
+ JsonNode root = objectMapper.readTree(responseBody);
+ if (root.has("error")) {
+ throw new SorobanRpcException("Soroban RPC rpcId=" + requestId + " method=" + method
+ + " error: " + truncate(root.get("error").toString()));
+ }
+ if (!root.has("result")) {
+ throw new SorobanRpcException("Soroban RPC rpcId=" + requestId + " method=" + method
+ + " response missing 'result'");
+ }
+ log.debug("Soroban RPC response rpcId={} method={} status={}", requestId, method,
+ textOrNull(root.get("result"), "status"));
+ return root.get("result");
+ } catch (IOException ex) {
+ // includes connect/read/write timeouts, which use the same callTimeout
+ // budget above rather than OkHttp's per-phase defaults, so a stuck
+ // Soroban RPC endpoint can't pin the calling thread indefinitely.
+ String message = "Soroban RPC rpcId=" + requestId + " method=" + method + " call failed: " + ex.getMessage();
+ log.warn(message);
+ throw new SorobanRpcException(message, ex);
+ }
+ }
+
+ private static String truncate(String value) {
+ if (value == null || value.length() <= MAX_LOGGED_BODY_LENGTH) {
+ return value;
+ }
+ return value.substring(0, MAX_LOGGED_BODY_LENGTH) + "…(truncated, " + value.length() + " chars total)";
+ }
+
+ private static String textOrNull(JsonNode node, String field) {
+ return node.hasNonNull(field) ? node.get(field).asText() : null;
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcException.java b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcException.java
new file mode 100644
index 0000000..86eee80
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcException.java
@@ -0,0 +1,12 @@
+package com.guildworkman.api.escrow.rpc;
+
+/** Raised for any transport or JSON-RPC-level failure talking to Soroban RPC. */
+public class SorobanRpcException extends RuntimeException {
+ public SorobanRpcException(String message) {
+ super(message);
+ }
+
+ public SorobanRpcException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcProperties.java b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcProperties.java
new file mode 100644
index 0000000..22faecc
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/rpc/SorobanRpcProperties.java
@@ -0,0 +1,26 @@
+package com.guildworkman.api.escrow.rpc;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+
+/**
+ * Binds the {@code soroban.rpc.*} properties used to reach a Soroban RPC
+ * endpoint (e.g. a Horizon/RPC provider or a local {@code soroban-rpc}
+ * instance) for submitting and polling escrow-contract transactions.
+ */
+@Component
+@ConfigurationProperties(prefix = "soroban.rpc")
+@Getter
+@Setter
+public class SorobanRpcProperties {
+
+ /** JSON-RPC endpoint, e.g. {@code https://soroban-testnet.stellar.org}. */
+ private String url = "https://soroban-testnet.stellar.org";
+
+ /** HTTP call timeout for each JSON-RPC request. */
+ private Duration requestTimeout = Duration.ofSeconds(10);
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationInserter.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationInserter.java
new file mode 100644
index 0000000..f049991
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationInserter.java
@@ -0,0 +1,61 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.escrow.api.SubmitOrchestrationRequest;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+
+/**
+ * Isolated insert so a unique-key race on {@code idempotency_key} aborts only
+ * this nested transaction (Postgres), leaving the caller's transaction able
+ * to re-read the winner. Mirrors {@code ChainEventInserter}.
+ *
+ * Why a unique constraint + {@code REQUIRES_NEW} instead of optimistic
+ * compare-and-swap? A CAS (read-then-conditional-insert-if-absent, or an
+ * application-level "check then insert") has a race window between the read
+ * and the write: two concurrent requests for the same new
+ * {@code idempotencyKey} can both observe "not present" and both attempt to
+ * insert, defeating the whole point of idempotency. The database's unique
+ * index is the only thing that can atomically decide a single winner across
+ * concurrent transactions. {@code REQUIRES_NEW} is what makes losing that
+ * race *cheap*: without it, the {@link org.springframework.dao.DataIntegrityViolationException}
+ * would poison the caller's own (potentially larger) transaction, forcing a
+ * full rollback of unrelated work just to read the winning row back.
+ *
+ *
Deadlocks / lock contention. This method only ever inserts a new
+ * row — Postgres does not need to acquire a row lock on anything already
+ * committed, so it cannot deadlock against {@link EscrowOrchestrationRequestRepository#claimNext}
+ * (which pessimistically locks *existing* {@code PENDING}/{@code SUBMITTED}
+ * rows). The only contention here is the unique index itself, which
+ * Postgres resolves by blocking the second inserter until the first commits
+ * or rolls back, then failing it with a unique-violation rather than
+ * deadlocking. Callers of {@link EscrowOrchestrationService#submit} don't
+ * need to retry on that failure themselves — it's caught and turned into a
+ * lookup of the winning row inside {@code submit} itself.
+ */
+@Service
+@RequiredArgsConstructor
+public class EscrowOrchestrationInserter {
+ private final EscrowOrchestrationRequestRepository repository;
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW)
+ public EscrowOrchestrationRequest insert(SubmitOrchestrationRequest request) {
+ EscrowOrchestrationRequest entity = new EscrowOrchestrationRequest();
+ entity.setIdempotencyKey(request.idempotencyKey());
+ entity.setOperationType(request.operationType());
+ entity.setContractId(request.contractId());
+ entity.setOperationRef(request.operationRef());
+ entity.setSignedTransactionXdr(request.signedTransactionXdr());
+ entity.setStatus(OrchestrationStatus.PENDING);
+ entity.setReconciliationStatus(ReconciliationStatus.PENDING);
+ entity.setNextAttemptAt(Instant.now());
+ return repository.saveAndFlush(entity);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationNotFoundException.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationNotFoundException.java
new file mode 100644
index 0000000..9aeac27
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationNotFoundException.java
@@ -0,0 +1,7 @@
+package com.guildworkman.api.escrow.service;
+
+public class EscrowOrchestrationNotFoundException extends RuntimeException {
+ public EscrowOrchestrationNotFoundException(Long id) {
+ super("Escrow orchestration request not found: " + id);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationRetryProperties.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationRetryProperties.java
new file mode 100644
index 0000000..e090cd2
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationRetryProperties.java
@@ -0,0 +1,38 @@
+package com.guildworkman.api.escrow.service;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+
+/**
+ * Binds the {@code escrow.orchestration.retry.*} properties governing
+ * {@link EscrowOrchestrationService}'s submit/poll retry loop. Exposed as
+ * config (rather than hardcoded constants) so both production tuning and
+ * tests can set deterministic values without recompiling.
+ */
+@Component
+@ConfigurationProperties(prefix = "escrow.orchestration.retry")
+@Getter
+@Setter
+public class EscrowOrchestrationRetryProperties {
+
+ /** Attempts (submit or poll, counted separately) before a request moves to DEAD_LETTER. */
+ private int maxAttempts = 5;
+
+ /** Backoff for attempt 1; doubles each subsequent attempt up to {@link #maxDelay}. */
+ private Duration baseDelay = Duration.ofSeconds(1);
+
+ /** Ceiling on the (pre-jitter) computed backoff, regardless of attempt count. */
+ private Duration maxDelay = Duration.ofSeconds(64);
+
+ /**
+ * Fraction of the computed backoff to randomize by, in both directions
+ * (e.g. {@code 0.2} spreads a 10s backoff over roughly [8s, 12s]) so many
+ * requests scheduled for the same instant don't all retry in lockstep
+ * against Soroban RPC.
+ */
+ private double jitter = 0.2;
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationService.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationService.java
new file mode 100644
index 0000000..c69883a
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowOrchestrationService.java
@@ -0,0 +1,269 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.escrow.api.SubmitOrchestrationRequest;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import com.guildworkman.api.escrow.rpc.GetTransactionResult;
+import com.guildworkman.api.escrow.rpc.SendTransactionResult;
+import com.guildworkman.api.escrow.rpc.SorobanRpcClient;
+import com.guildworkman.api.escrow.rpc.SorobanRpcException;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.EnumSet;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Submits and confirms escrow-contract operations over Soroban RPC.
+ *
+ *
Idempotency — {@link #submit} dedupes on {@code idempotencyKey}:
+ * resubmitting the same key returns the original request instead of creating
+ * a second one, so a client-side retry of the REST call never double-submits.
+ * See {@link EscrowOrchestrationInserter} for why the insert itself needs a
+ * nested transaction, and how that compares to an optimistic
+ * compare-and-swap.
+ *
+ *
Exactly-once — is achieved compositely, not by any single lock:
+ *
+ * - the idempotency key stops duplicate rows being created for the same
+ * logical request;
+ * - {@link #submitPending()} only ever claims {@code PENDING} rows (via a
+ * {@code SELECT ... FOR UPDATE}-backed query), so a given row's signed
+ * envelope is handed to {@code sendTransaction} at most once per
+ * process attempt;
+ * - even if a crash happens between the RPC call succeeding and the row
+ * being committed, Soroban RPC itself dedupes by the envelope's own
+ * hash — resubmitting identical XDR comes back {@code DUPLICATE} with
+ * the same hash rather than executing twice.
+ *
+ *
+ * Backoff — retry delay doubles from {@code retryProperties.baseDelay}
+ * each attempt up to {@code retryProperties.maxDelay}, then is randomized by
+ * {@code +/- retryProperties.jitter} so a batch of requests that failed
+ * together don't all retry in the same instant and hammer Soroban RPC again
+ * (a thundering herd). A request that exhausts {@code retryProperties.maxAttempts}
+ * moves to {@link OrchestrationStatus#DEAD_LETTER} — an operator needs to
+ * look at {@code lastError} and either fix the underlying cause and requeue
+ * it, or discard it.
+ */
+@Service
+@RequiredArgsConstructor
+public class EscrowOrchestrationService {
+
+ private static final Logger log = LoggerFactory.getLogger(EscrowOrchestrationService.class);
+
+ private final EscrowOrchestrationRequestRepository repository;
+ private final EscrowOrchestrationInserter inserter;
+ private final SorobanRpcClient sorobanRpcClient;
+ private final EscrowOrchestrationRetryProperties retryProperties;
+
+ /**
+ * @return the (possibly pre-existing) request for {@code request.idempotencyKey()},
+ * tagged with whether this call created it or returned an existing one
+ * (see {@link SubmitOutcome#replayed()}). Callers that only need the entity
+ * can use {@link SubmitOutcome#request()}.
+ */
+ @Transactional(readOnly = true)
+ public SubmitOutcome submit(SubmitOrchestrationRequest request) {
+ return repository.findByIdempotencyKey(request.idempotencyKey())
+ .map(existing -> new SubmitOutcome(existing, true))
+ .orElseGet(() -> insertIdempotently(request));
+ }
+
+ /** @param replayed true if {@code request} already existed for this idempotency key. */
+ public record SubmitOutcome(EscrowOrchestrationRequest request, boolean replayed) {
+ }
+
+ @Transactional(readOnly = true)
+ public EscrowOrchestrationRequest get(Long id) {
+ return repository.findById(id).orElseThrow(() -> new EscrowOrchestrationNotFoundException(id));
+ }
+
+ /**
+ * Resets a {@link com.guildworkman.api.escrow.model.ReconciliationStatus#MISMATCHED}
+ * request back to {@code PENDING} so {@link EscrowReconciliationService#reconcilePending()}
+ * reconsiders it on its next sweep — e.g. after confirming the missing
+ * on-chain event has now been ingested. Wraps the manual SQL documented
+ * in {@code docs/ESCROW_ORCHESTRATION.md} ("Operations") in an
+ * ADMIN-gated endpoint; see {@code EscrowOrchestrationController}.
+ */
+ @Transactional
+ public EscrowOrchestrationRequest requeueReconciliation(Long id) {
+ EscrowOrchestrationRequest entity = repository.findById(id)
+ .orElseThrow(() -> new EscrowOrchestrationNotFoundException(id));
+ if (entity.getReconciliationStatus() != ReconciliationStatus.MISMATCHED) {
+ throw new ReconciliationRequeueNotAllowedException(id, entity.getReconciliationStatus());
+ }
+ entity.setReconciliationStatus(ReconciliationStatus.PENDING);
+ entity.setReconciledAt(null);
+ repository.save(entity);
+ log.info("Escrow orchestration request id={} reconciliation requeued (was MISMATCHED)", id);
+ return entity;
+ }
+
+ private SubmitOutcome insertIdempotently(SubmitOrchestrationRequest request) {
+ try {
+ EscrowOrchestrationRequest created = inserter.insert(request);
+ log.info("Escrow orchestration request created id={} operationType={} operationRef={}",
+ created.getId(), created.getOperationType(), created.getOperationRef());
+ return new SubmitOutcome(created, false);
+ } catch (DataIntegrityViolationException ex) {
+ // Lost the unique-key race: another concurrent call for the same
+ // idempotency key won. The nested REQUIRES_NEW insert rolled back
+ // on its own, so the outer (read-only) transaction can still read
+ // the winner — this call is a replay too, just one that raced.
+ EscrowOrchestrationRequest winner = repository.findByIdempotencyKey(request.idempotencyKey())
+ .orElseThrow(() -> new IllegalStateException(
+ "Orchestration request not found after idempotent-guard violation for key="
+ + request.idempotencyKey(), ex));
+ return new SubmitOutcome(winner, true);
+ }
+ }
+
+ @Scheduled(fixedDelayString = "${escrow.orchestration.submit-poll-delay-ms:1000}")
+ @Transactional
+ public void submitPending() {
+ repository.claimNext(EnumSet.of(OrchestrationStatus.PENDING), Instant.now(), PageRequest.of(0, 1))
+ .stream().findFirst().ifPresent(this::submitOne);
+ }
+
+ void submitOne(EscrowOrchestrationRequest entity) {
+ entity.setAttempts(entity.getAttempts() + 1);
+ try {
+ SendTransactionResult result = sorobanRpcClient.sendTransaction(entity.getSignedTransactionXdr());
+ if (result.isAccepted()) {
+ entity.setSorobanTxHash(result.hash());
+ entity.setStatus(OrchestrationStatus.SUBMITTED);
+ entity.setSubmittedAt(Instant.now());
+ entity.setNextAttemptAt(Instant.now());
+ entity.setLastError(null);
+ repository.save(entity);
+ log.info("Escrow orchestration request id={} submitted sorobanTxHash={} attempts={}",
+ entity.getId(), entity.getSorobanTxHash(), entity.getAttempts());
+ } else if (result.isRetryable()) {
+ scheduleRetry(entity, "Soroban RPC busy (TRY_AGAIN_LATER)");
+ } else {
+ fail(entity, "Soroban RPC rejected transaction: " + result.errorResultXdr());
+ }
+ } catch (SorobanRpcException ex) {
+ scheduleRetry(entity, ex.getMessage());
+ }
+ }
+
+ @Scheduled(fixedDelayString = "${escrow.orchestration.confirm-poll-delay-ms:1000}")
+ @Transactional
+ public void pollSubmitted() {
+ repository.claimNext(EnumSet.of(OrchestrationStatus.SUBMITTED), Instant.now(), PageRequest.of(0, 1))
+ .stream().findFirst().ifPresent(this::pollOne);
+ }
+
+ void pollOne(EscrowOrchestrationRequest entity) {
+ try {
+ GetTransactionResult result = sorobanRpcClient.getTransaction(entity.getSorobanTxHash());
+ if (result.isSuccess()) {
+ entity.setStatus(OrchestrationStatus.CONFIRMED);
+ entity.setConfirmedAt(Instant.now());
+ entity.setLastError(null);
+ repository.save(entity);
+ log.info("Escrow orchestration request id={} confirmed sorobanTxHash={} ledger={}",
+ entity.getId(), entity.getSorobanTxHash(), result.ledger());
+ } else if (result.isFailed()) {
+ // The on-chain transaction was rejected; the signed envelope's
+ // sequence number is consumed, so resubmitting it can never
+ // succeed. Terminal, not retried.
+ entity.setStatus(OrchestrationStatus.FAILED);
+ entity.setLastError("Soroban transaction failed on-chain: " + result.resultXdr());
+ repository.save(entity);
+ log.warn("Escrow orchestration request id={} failed on-chain sorobanTxHash={}",
+ entity.getId(), entity.getSorobanTxHash());
+ } else {
+ // NOT_FOUND: not yet ingested by RPC's ledger view. Keep polling.
+ entity.setAttempts(entity.getAttempts() + 1);
+ if (entity.getAttempts() >= retryProperties.getMaxAttempts()) {
+ deadLetter(entity, "Gave up waiting for transaction confirmation after "
+ + retryProperties.getMaxAttempts() + " polls");
+ } else {
+ entity.setNextAttemptAt(nextAttemptAt(entity.getAttempts()));
+ repository.save(entity);
+ }
+ }
+ } catch (SorobanRpcException ex) {
+ log.warn("Soroban RPC poll failed for orchestration id={} errorClass={}: {}",
+ entity.getId(), ex.getClass().getSimpleName(), ex.getMessage());
+ entity.setAttempts(entity.getAttempts() + 1);
+ entity.setLastError(ex.getMessage());
+ if (entity.getAttempts() >= retryProperties.getMaxAttempts()) {
+ deadLetter(entity, ex.getMessage());
+ } else {
+ entity.setNextAttemptAt(nextAttemptAt(entity.getAttempts()));
+ repository.save(entity);
+ }
+ }
+ }
+
+ private void scheduleRetry(EscrowOrchestrationRequest entity, String error) {
+ if (entity.getAttempts() >= retryProperties.getMaxAttempts()) {
+ deadLetter(entity, error);
+ return;
+ }
+ entity.setLastError(error);
+ entity.setStatus(OrchestrationStatus.PENDING);
+ entity.setNextAttemptAt(nextAttemptAt(entity.getAttempts()));
+ repository.save(entity);
+ log.info("Escrow orchestration request id={} retry scheduled attempts={} nextAttemptAt={} cause={}",
+ entity.getId(), entity.getAttempts(), entity.getNextAttemptAt(), error);
+ }
+
+ private void fail(EscrowOrchestrationRequest entity, String error) {
+ entity.setStatus(OrchestrationStatus.FAILED);
+ entity.setLastError(error);
+ repository.save(entity);
+ log.warn("Escrow orchestration request id={} failed: {}", entity.getId(), error);
+ }
+
+ /**
+ * Terminal: persists the entity itself (unlike {@link #fail}/{@link #scheduleRetry}'s
+ * non-terminal branches, this has no caller that does anything further with
+ * {@code entity} afterward, so saving here — rather than requiring every
+ * call site to remember to — removes a class of "forgot to persist"
+ * bugs). {@code nextAttemptAt} is left at its last computed value rather
+ * than cleared: it's inert once {@code DEAD_LETTER}, since
+ * {@link EscrowOrchestrationRequestRepository#claimNext} only ever
+ * selects {@code PENDING}/{@code SUBMITTED} rows, and keeping it records
+ * "when the next attempt would have been" for diagnostics.
+ */
+ private void deadLetter(EscrowOrchestrationRequest entity, String error) {
+ entity.setStatus(OrchestrationStatus.DEAD_LETTER);
+ entity.setLastError(error);
+ repository.save(entity);
+ log.warn("Escrow orchestration request id={} moved to DEAD_LETTER after {} attempts: {}",
+ entity.getId(), entity.getAttempts(), error);
+ }
+
+ /**
+ * Backoff doubles from {@code baseDelay} per attempt, capped at
+ * {@code maxDelay}, then jittered by {@code +/- jitter} (a fraction of the
+ * capped delay) to avoid a thundering herd of retries hitting Soroban RPC
+ * at the same instant.
+ */
+ private Instant nextAttemptAt(int attempts) {
+ long baseMillis = Math.max(1, retryProperties.getBaseDelay().toMillis());
+ long capMillis = Math.max(baseMillis, retryProperties.getMaxDelay().toMillis());
+ int shift = Math.min(Math.max(attempts - 1, 0), 20); // guards against overflow on shift
+ long raw = baseMillis << shift;
+ long capped = (raw < 0 || raw > capMillis) ? capMillis : raw; // raw < 0 means it overflowed
+ double jitter = Math.max(0, retryProperties.getJitter());
+ double randomOffset = jitter <= 0 ? 0 : ThreadLocalRandom.current().nextDouble(-jitter, jitter);
+ long delayMillis = Math.max(1, Math.round(capped * (1.0 + randomOffset)));
+ return Instant.now().plusMillis(delayMillis);
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationProperties.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationProperties.java
new file mode 100644
index 0000000..0e03c5d
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationProperties.java
@@ -0,0 +1,22 @@
+package com.guildworkman.api.escrow.service;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+
+@Component
+@ConfigurationProperties(prefix = "escrow.reconciliation")
+@Getter
+@Setter
+public class EscrowReconciliationProperties {
+
+ /**
+ * How long a CONFIRMED request may go without a corroborating on-chain
+ * event before it's flagged MISMATCHED. Gives the ingestion indexer
+ * (see {@code com.guildworkman.api.chain}) time to catch up.
+ */
+ private Duration window = Duration.ofMinutes(10);
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationService.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationService.java
new file mode 100644
index 0000000..ba33a3d
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/EscrowReconciliationService.java
@@ -0,0 +1,94 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.chain.model.ChainEventStatus;
+import com.guildworkman.api.chain.model.OnChainEvent;
+import com.guildworkman.api.chain.repository.OnChainEventRepository;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * Reconciles off-chain orchestration state against on-chain reality.
+ *
+ *
Rather than issuing our own Soroban RPC ledger-entry reads (which would
+ * require encoding {@code LedgerKey}/{@code ScVal} XDR by hand — the very
+ * thing {@link com.guildworkman.api.escrow.rpc.SorobanRpcClient} deliberately
+ * avoids), reconciliation is done against the on-chain event stream already
+ * ingested by {@code com.guildworkman.api.chain} (see issue #22's
+ * transactional-outbox ingestion pipeline). A {@link OrchestrationStatus#CONFIRMED}
+ * request is considered corroborated once a {@link ChainEventStatus#PROCESSED}
+ * event for the same contract, tagged with this request's {@code operationRef}
+ * as one of its topics, has been ingested.
+ *
+ *
By convention, the indexer feeding {@code /api/v1/chain/events} is
+ * expected to include the affected appointment/escrow id as an event topic
+ * — topics already model semantic Soroban event tags (see
+ * {@code IngestChainEventRequest}).
+ */
+@Service
+@RequiredArgsConstructor
+public class EscrowReconciliationService {
+
+ private static final int BATCH_SIZE = 20;
+
+ private final EscrowOrchestrationRequestRepository orchestrationRequests;
+ private final OnChainEventRepository onChainEvents;
+ private final EscrowReconciliationProperties properties;
+
+ @Scheduled(fixedDelayString = "${escrow.reconciliation.poll-delay-ms:5000}")
+ @Transactional
+ public void reconcilePending() {
+ List candidates = orchestrationRequests.findByStatusAndReconciliationStatus(
+ OrchestrationStatus.CONFIRMED, ReconciliationStatus.PENDING, PageRequest.of(0, BATCH_SIZE));
+ candidates.forEach(this::reconcileOne);
+ }
+
+ void reconcileOne(EscrowOrchestrationRequest entity) {
+ // OnChainEvent.topics is a JSON-encoded string column (see
+ // ChainEventInserter), not a normalized list, so matching a specific
+ // topic value means matching its literal JSON-quoted form as a
+ // substring — e.g. operationRef "42" must appear as `"42"` inside
+ // something like `["42","Completed"]`. This is a LIKE %..% query
+ // (OnChainEventRepository#findByContractIdAndTopicsContaining); the
+ // quotes keep it from misfiring on a numeric prefix ("4" won't match
+ // "42"). SubmitOrchestrationRequest rejects a literal '"' in
+ // operationRef so it can't break out of this quoting.
+ String topicFragment = "\"" + entity.getOperationRef() + "\"";
+ List matches = onChainEvents.findByContractIdAndTopicsContaining(
+ entity.getContractId(), topicFragment);
+
+ // PROCESSED (not just "ingested") because a PENDING/PROCESSING event
+ // might still fail its own retries and never actually reflect this
+ // operation having taken effect — see ChainEventService.
+ boolean corroborated = matches.stream().anyMatch(e -> e.getStatus() == ChainEventStatus.PROCESSED);
+ if (corroborated) {
+ entity.setReconciliationStatus(ReconciliationStatus.MATCHED);
+ entity.setReconciledAt(Instant.now());
+ orchestrationRequests.save(entity);
+ return;
+ }
+
+ Instant deadline = entity.getConfirmedAt() != null
+ ? entity.getConfirmedAt().plus(properties.getWindow())
+ : Instant.now();
+ if (Instant.now().isAfter(deadline)) {
+ // Terminal: MISMATCHED is not automatically retried by this sweep
+ // (findByStatusAndReconciliationStatus only selects PENDING). See
+ // docs/ESCROW_ORCHESTRATION.md "Operations" for how to force a
+ // recheck once the indexer catches up.
+ entity.setReconciliationStatus(ReconciliationStatus.MISMATCHED);
+ entity.setReconciledAt(Instant.now());
+ orchestrationRequests.save(entity);
+ }
+ // else: still within the grace window — leave PENDING for the next tick.
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/escrow/service/ReconciliationRequeueNotAllowedException.java b/backend-api/src/main/java/com/guildworkman/api/escrow/service/ReconciliationRequeueNotAllowedException.java
new file mode 100644
index 0000000..d39feab
--- /dev/null
+++ b/backend-api/src/main/java/com/guildworkman/api/escrow/service/ReconciliationRequeueNotAllowedException.java
@@ -0,0 +1,11 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+
+/** Only a {@link ReconciliationStatus#MISMATCHED} request can be requeued. */
+public class ReconciliationRequeueNotAllowedException extends RuntimeException {
+ public ReconciliationRequeueNotAllowedException(Long id, ReconciliationStatus current) {
+ super("Escrow orchestration request " + id + " is not MISMATCHED (current: " + current
+ + "); only MISMATCHED requests can be requeued for reconciliation");
+ }
+}
diff --git a/backend-api/src/main/java/com/guildworkman/api/handler/GlobalExceptionHandler.java b/backend-api/src/main/java/com/guildworkman/api/handler/GlobalExceptionHandler.java
index b6e57ef..bde7495 100644
--- a/backend-api/src/main/java/com/guildworkman/api/handler/GlobalExceptionHandler.java
+++ b/backend-api/src/main/java/com/guildworkman/api/handler/GlobalExceptionHandler.java
@@ -1,5 +1,7 @@
package com.guildworkman.api.handler;
+import com.guildworkman.api.escrow.service.EscrowOrchestrationNotFoundException;
+import com.guildworkman.api.escrow.service.ReconciliationRequeueNotAllowedException;
import com.guildworkman.api.exceptions.*;
import jakarta.validation.ConstraintViolationException;
import org.slf4j.Logger;
@@ -89,6 +91,18 @@ public ResponseEntity handleAppointmentNotFound(AppointmentNotFou
"Appointment not found", exception.getMessage());
}
+ @ExceptionHandler(EscrowOrchestrationNotFoundException.class)
+ public ResponseEntity handleEscrowOrchestrationNotFound(EscrowOrchestrationNotFoundException exception) {
+ return respond(HttpStatus.NOT_FOUND, "escrow-orchestration-not-found",
+ "Escrow orchestration request not found", exception.getMessage());
+ }
+
+ @ExceptionHandler(ReconciliationRequeueNotAllowedException.class)
+ public ResponseEntity handleReconciliationRequeueNotAllowed(ReconciliationRequeueNotAllowedException exception) {
+ return respond(HttpStatus.CONFLICT, "reconciliation-requeue-not-allowed",
+ "Reconciliation requeue not allowed", exception.getMessage());
+ }
+
@ExceptionHandler(InvalidPasswordException.class)
public ResponseEntity handleInvalidPasswordException(InvalidPasswordException exception) {
return respond(HttpStatus.BAD_REQUEST, "invalid-password",
diff --git a/backend-api/src/main/resources/application.properties b/backend-api/src/main/resources/application.properties
index 918a5c2..d8c78dc 100644
--- a/backend-api/src/main/resources/application.properties
+++ b/backend-api/src/main/resources/application.properties
@@ -2,6 +2,16 @@ spring.application.name=guildworkman-api
spring.data.jdbc.dialect=postgresql
spring.datasource.driver-class-name=org.postgresql.Driver
spring.jpa.database-platform=org.hibernate.dialect.PostgreSQLDialect
+# Every table/constraint in this app (including escrow_orchestration_requests
+# and on_chain_events) is schema-managed this way — there is no Flyway/
+# Liquibase here. `update` creates missing tables/constraints on startup and
+# never drops or destructively alters existing columns, but it also can't
+# express a safe rename or type change; those still need a hand-written
+# ALTER TABLE. Production environments that want migration-tracked, reviewable
+# schema changes should consider introducing Flyway/Liquibase — that's a
+# cross-cutting change affecting every table, not specific to any one feature.
+# See backend-api/docs/ESCROW_ORCHESTRATION.md ("Schema / migrations") for
+# more detail.
spring.jpa.hibernate.ddl-auto=${DDL_AUTO:update}
spring.datasource.url=${DATABASE_URL:jdbc:postgresql://localhost:5432/service}
@@ -36,6 +46,30 @@ guildworkman.problem-details.type-base=${PROBLEM_TYPE_BASE:https://guildworkman.
# On-chain ingestion worker. Set to 0 to disable polling in a batch/replay job.
chain.events.poll-delay-ms=${CHAIN_EVENTS_POLL_DELAY_MS:1000}
+# Escrow orchestration: Soroban RPC endpoint used to submit/poll escrow
+# contract transactions. Callers build and sign transactions client-side;
+# this service only relays the signed XDR and tracks its outcome.
+soroban.rpc.url=${SOROBAN_RPC_URL:https://soroban-testnet.stellar.org}
+soroban.rpc.request-timeout=${SOROBAN_RPC_TIMEOUT:PT10S}
+
+# How often the orchestration workers poll for PENDING/SUBMITTED requests.
+escrow.orchestration.submit-poll-delay-ms=${ESCROW_SUBMIT_POLL_DELAY_MS:1000}
+escrow.orchestration.confirm-poll-delay-ms=${ESCROW_CONFIRM_POLL_DELAY_MS:1000}
+
+# Retry/backoff for both the submit and confirm loops. Backoff doubles from
+# base-delay each attempt, capped at max-delay, then randomized by +/- jitter
+# (a fraction of the computed delay) so retries don't all land in lockstep.
+escrow.orchestration.retry.max-attempts=${ESCROW_RETRY_MAX_ATTEMPTS:5}
+escrow.orchestration.retry.base-delay=${ESCROW_RETRY_BASE_DELAY:PT1S}
+escrow.orchestration.retry.max-delay=${ESCROW_RETRY_MAX_DELAY:PT64S}
+escrow.orchestration.retry.jitter=${ESCROW_RETRY_JITTER:0.2}
+
+# Reconciliation: how long a CONFIRMED request may go without a corroborating
+# ingested on-chain event before it's flagged MISMATCHED, and how often the
+# reconciliation sweep runs.
+escrow.reconciliation.window=${ESCROW_RECONCILIATION_WINDOW:PT10M}
+escrow.reconciliation.poll-delay-ms=${ESCROW_RECONCILIATION_POLL_DELAY_MS:5000}
+
# JWT authentication.
# - secret: HMAC-256 signing key. MUST be overridden in every real
# environment via JWT_SECRET (>= 32 bytes). The baked-in default exists
diff --git a/backend-api/src/test/java/com/guildworkman/api/escrow/EscrowOrchestrationIntegrationTest.java b/backend-api/src/test/java/com/guildworkman/api/escrow/EscrowOrchestrationIntegrationTest.java
new file mode 100644
index 0000000..3f4d97d
--- /dev/null
+++ b/backend-api/src/test/java/com/guildworkman/api/escrow/EscrowOrchestrationIntegrationTest.java
@@ -0,0 +1,269 @@
+package com.guildworkman.api.escrow;
+
+import com.guildworkman.api.chain.model.ChainEventStatus;
+import com.guildworkman.api.chain.model.OnChainEvent;
+import com.guildworkman.api.chain.repository.OnChainEventRepository;
+import com.guildworkman.api.escrow.api.SubmitOrchestrationRequest;
+import com.guildworkman.api.escrow.model.EscrowOperationType;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import com.guildworkman.api.escrow.rpc.GetTransactionResult;
+import com.guildworkman.api.escrow.rpc.SendTransactionResult;
+import com.guildworkman.api.escrow.rpc.SorobanRpcClient;
+import com.guildworkman.api.escrow.rpc.SorobanRpcException;
+import com.guildworkman.api.escrow.service.EscrowOrchestrationService;
+import com.guildworkman.api.escrow.service.EscrowReconciliationService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.when;
+
+@SpringBootTest(properties = {
+ "escrow.orchestration.submit-poll-delay-ms=3600000",
+ "escrow.orchestration.confirm-poll-delay-ms=3600000",
+ "escrow.reconciliation.poll-delay-ms=3600000",
+ "chain.events.poll-delay-ms=3600000",
+ "spring.task.scheduling.enabled=false"
+})
+class EscrowOrchestrationIntegrationTest {
+
+ @Autowired
+ private EscrowOrchestrationRequestRepository orchestrationRequests;
+
+ @Autowired
+ private OnChainEventRepository onChainEvents;
+
+ @Autowired
+ private EscrowOrchestrationService orchestrationService;
+
+ @Autowired
+ private EscrowReconciliationService reconciliationService;
+
+ @MockBean
+ private SorobanRpcClient sorobanRpcClient;
+
+ @BeforeEach
+ void cleanSlate() {
+ reset(sorobanRpcClient);
+ orchestrationRequests.deleteAll();
+ onChainEvents.deleteAll();
+ }
+
+ private static String key(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ private static SubmitOrchestrationRequest submitRequest(String idemKey, String operationRef) {
+ return new SubmitOrchestrationRequest(idemKey, EscrowOperationType.CONFIRM_COMPLETION, "CTEST", operationRef, "AAAA==");
+ }
+
+ // --- submit() idempotency -----------------------------------------------
+
+ @Test
+ void submitCreatesAPendingRequest() {
+ var outcome = orchestrationService.submit(submitRequest(key("k1"), "1"));
+ assertThat(outcome.replayed()).isFalse();
+
+ var saved = orchestrationRequests.findById(outcome.request().getId()).orElseThrow();
+ assertThat(saved.getStatus()).isEqualTo(OrchestrationStatus.PENDING);
+ assertThat(saved.getReconciliationStatus()).isEqualTo(ReconciliationStatus.PENDING);
+ }
+
+ @Test
+ void submitIsIdempotentForDuplicateKey() {
+ String idemKey = key("dup");
+ var first = orchestrationService.submit(submitRequest(idemKey, "1"));
+ var second = orchestrationService.submit(submitRequest(idemKey, "1"));
+
+ assertThat(first.replayed()).isFalse();
+ assertThat(second.replayed()).isTrue();
+ assertThat(second.request().getId()).isEqualTo(first.request().getId());
+ assertThat(orchestrationRequests.count()).isEqualTo(1);
+ }
+
+ @Test
+ void concurrentSubmitsWithSameIdempotencyKeyCreateOnlyOneRow() throws Exception {
+ String idemKey = key("race");
+ int threads = 8;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ List> tasks = new java.util.ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ tasks.add(() -> orchestrationService.submit(submitRequest(idemKey, "1")).request().getId());
+ }
+
+ List> futures = pool.invokeAll(tasks, 30, TimeUnit.SECONDS);
+ List ids = new java.util.ArrayList<>();
+ for (Future f : futures) {
+ ids.add(f.get(5, TimeUnit.SECONDS));
+ }
+ pool.shutdown();
+ assertThat(pool.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+
+ assertThat(ids.stream().distinct().count()).isEqualTo(1);
+ assertThat(orchestrationRequests.count()).isEqualTo(1);
+ }
+
+ // --- full submit -> confirm lifecycle -----------------------------------
+
+ @Test
+ void fullLifecycleFromSubmitToConfirmed() {
+ var resp = orchestrationService.submit(submitRequest(key("lifecycle"), "7")).request();
+ when(sorobanRpcClient.sendTransaction(any())).thenReturn(new SendTransactionResult("hash-1", "PENDING", null));
+
+ orchestrationService.submitPending();
+ EscrowOrchestrationRequest afterSubmit = orchestrationRequests.findById(resp.getId()).orElseThrow();
+ assertThat(afterSubmit.getStatus()).isEqualTo(OrchestrationStatus.SUBMITTED);
+ assertThat(afterSubmit.getSorobanTxHash()).isEqualTo("hash-1");
+
+ when(sorobanRpcClient.getTransaction("hash-1")).thenReturn(new GetTransactionResult("SUCCESS", "xdr", 100L));
+ orchestrationService.pollSubmitted();
+
+ EscrowOrchestrationRequest confirmed = orchestrationRequests.findById(resp.getId()).orElseThrow();
+ assertThat(confirmed.getStatus()).isEqualTo(OrchestrationStatus.CONFIRMED);
+ assertThat(confirmed.getConfirmedAt()).isNotNull();
+ }
+
+ // Mirrors EscrowOrchestrationRetryProperties' default maxAttempts (this
+ // test doesn't override escrow.orchestration.retry.max-attempts).
+ private static final int MAX_ATTEMPTS = 5;
+
+ @Test
+ void rpcFailureAppliesBackoffThenDeadLetter() {
+ var resp = orchestrationService.submit(submitRequest(key("dl"), "9")).request();
+ // Force nextAttemptAt into the past and attempts near the ceiling so the
+ // very next failure trips DEAD_LETTER without looping through backoff delays.
+ EscrowOrchestrationRequest entity = orchestrationRequests.findById(resp.getId()).orElseThrow();
+ entity.setAttempts(MAX_ATTEMPTS - 1);
+ entity.setNextAttemptAt(Instant.now().minusSeconds(1));
+ orchestrationRequests.saveAndFlush(entity);
+
+ when(sorobanRpcClient.sendTransaction(any())).thenThrow(new SorobanRpcException("boom"));
+ orchestrationService.submitPending();
+
+ EscrowOrchestrationRequest reloaded = orchestrationRequests.findById(resp.getId()).orElseThrow();
+ assertThat(reloaded.getStatus()).isEqualTo(OrchestrationStatus.DEAD_LETTER);
+ assertThat(reloaded.getAttempts()).isEqualTo(MAX_ATTEMPTS);
+ }
+
+ @Test
+ void pessimisticLockingPreventsDoubleSubmission() throws Exception {
+ var resp = orchestrationService.submit(submitRequest(key("concurrent"), "3")).request();
+ when(sorobanRpcClient.sendTransaction(any())).thenReturn(new SendTransactionResult("hash-x", "PENDING", null));
+
+ int threads = 8;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ List> tasks = new java.util.ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ tasks.add(() -> {
+ orchestrationService.submitPending();
+ return null;
+ });
+ }
+ List> futures = pool.invokeAll(tasks, 30, TimeUnit.SECONDS);
+ for (Future f : futures) {
+ f.get(5, TimeUnit.SECONDS);
+ }
+ pool.shutdown();
+ assertThat(pool.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+
+ EscrowOrchestrationRequest reloaded = orchestrationRequests.findById(resp.getId()).orElseThrow();
+ assertThat(reloaded.getStatus()).isEqualTo(OrchestrationStatus.SUBMITTED);
+ assertThat(reloaded.getAttempts()).isEqualTo(1);
+ }
+
+ // --- reconciliation -------------------------------------------------------
+
+ private EscrowOrchestrationRequest saveConfirmed(String operationRef, Instant confirmedAt) {
+ EscrowOrchestrationRequest e = new EscrowOrchestrationRequest();
+ e.setIdempotencyKey(key("recon"));
+ e.setOperationType(EscrowOperationType.CONFIRM_COMPLETION);
+ e.setContractId("CTEST");
+ e.setOperationRef(operationRef);
+ e.setSignedTransactionXdr("AAAA==");
+ e.setStatus(OrchestrationStatus.CONFIRMED);
+ e.setReconciliationStatus(ReconciliationStatus.PENDING);
+ e.setConfirmedAt(confirmedAt);
+ e.setNextAttemptAt(Instant.now());
+ return orchestrationRequests.saveAndFlush(e);
+ }
+
+ private void saveChainEvent(String operationRef, ChainEventStatus status) {
+ OnChainEvent event = new OnChainEvent();
+ event.setEventKey(key("evt"));
+ event.setContractId("CTEST");
+ event.setLedger(1);
+ event.setEventIndex(0);
+ event.setTopics("[\"" + operationRef + "\"]");
+ event.setPayload("{}");
+ event.setStatus(status);
+ event.setNextAttemptAt(Instant.now());
+ onChainEvents.saveAndFlush(event);
+ }
+
+ @Test
+ void reconciliationMatchesWhenProcessedEventExists() {
+ var request = saveConfirmed("55", Instant.now());
+ saveChainEvent("55", ChainEventStatus.PROCESSED);
+
+ reconciliationService.reconcilePending();
+
+ EscrowOrchestrationRequest reloaded = orchestrationRequests.findById(request.getId()).orElseThrow();
+ assertThat(reloaded.getReconciliationStatus()).isEqualTo(ReconciliationStatus.MATCHED);
+ }
+
+ @Test
+ void reconciliationFlagsMismatchAfterWindowElapsesWithNoEvent() {
+ var request = saveConfirmed("66", Instant.now().minusSeconds(20 * 60));
+
+ reconciliationService.reconcilePending();
+
+ EscrowOrchestrationRequest reloaded = orchestrationRequests.findById(request.getId()).orElseThrow();
+ assertThat(reloaded.getReconciliationStatus()).isEqualTo(ReconciliationStatus.MISMATCHED);
+ }
+
+ @Test
+ void reconciliationLeavesPendingWithinGraceWindow() {
+ var request = saveConfirmed("77", Instant.now());
+
+ reconciliationService.reconcilePending();
+
+ EscrowOrchestrationRequest reloaded = orchestrationRequests.findById(request.getId()).orElseThrow();
+ assertThat(reloaded.getReconciliationStatus()).isEqualTo(ReconciliationStatus.PENDING);
+ }
+
+ @Test
+ void requeuedMismatchIsPickedUpByTheNextReconciliationSweep() {
+ var request = saveConfirmed("88", Instant.now().minusSeconds(20 * 60));
+ reconciliationService.reconcilePending();
+ assertThat(orchestrationRequests.findById(request.getId()).orElseThrow().getReconciliationStatus())
+ .isEqualTo(ReconciliationStatus.MISMATCHED);
+
+ orchestrationService.requeueReconciliation(request.getId());
+ assertThat(orchestrationRequests.findById(request.getId()).orElseThrow().getReconciliationStatus())
+ .isEqualTo(ReconciliationStatus.PENDING);
+
+ // Now the corroborating event shows up before the sweep runs again.
+ saveChainEvent("88", ChainEventStatus.PROCESSED);
+ reconciliationService.reconcilePending();
+
+ assertThat(orchestrationRequests.findById(request.getId()).orElseThrow().getReconciliationStatus())
+ .isEqualTo(ReconciliationStatus.MATCHED);
+ }
+}
diff --git a/backend-api/src/test/java/com/guildworkman/api/escrow/api/SubmitOrchestrationRequestValidationTest.java b/backend-api/src/test/java/com/guildworkman/api/escrow/api/SubmitOrchestrationRequestValidationTest.java
new file mode 100644
index 0000000..3614930
--- /dev/null
+++ b/backend-api/src/test/java/com/guildworkman/api/escrow/api/SubmitOrchestrationRequestValidationTest.java
@@ -0,0 +1,76 @@
+package com.guildworkman.api.escrow.api;
+
+import com.guildworkman.api.escrow.model.EscrowOperationType;
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class SubmitOrchestrationRequestValidationTest {
+
+ private static ValidatorFactory factory;
+ private static Validator validator;
+
+ @BeforeAll
+ static void setUp() {
+ factory = Validation.buildDefaultValidatorFactory();
+ validator = factory.getValidator();
+ }
+
+ @AfterAll
+ static void tearDown() {
+ factory.close();
+ }
+
+ private static SubmitOrchestrationRequest request(String signedTransactionXdr) {
+ return new SubmitOrchestrationRequest(
+ "idem-1", EscrowOperationType.CONFIRM_COMPLETION, "CABC", "42", signedTransactionXdr);
+ }
+
+ @Test
+ void acceptsAValidBase64Envelope() {
+ Set> violations = validator.validate(request("AAAA=="));
+ assertThat(violations).isEmpty();
+ }
+
+ @Test
+ void rejectsAnOversizedEnvelope() {
+ String oversized = "A".repeat(8193);
+ Set> violations = validator.validate(request(oversized));
+ assertThat(violations).anyMatch(v -> v.getPropertyPath().toString().equals("signedTransactionXdr"));
+ }
+
+ @Test
+ void acceptsAnEnvelopeAtTheSizeLimit() {
+ String atLimit = "A".repeat(8192);
+ Set> violations = validator.validate(request(atLimit));
+ assertThat(violations).isEmpty();
+ }
+
+ @Test
+ void rejectsNonBase64Content() {
+ Set> violations = validator.validate(request("not valid xdr!!"));
+ assertThat(violations).anyMatch(v -> v.getPropertyPath().toString().equals("signedTransactionXdr"));
+ }
+
+ @Test
+ void rejectsBlankEnvelope() {
+ Set> violations = validator.validate(request(""));
+ assertThat(violations).anyMatch(v -> v.getPropertyPath().toString().equals("signedTransactionXdr"));
+ }
+
+ @Test
+ void rejectsOperationRefContainingAQuote() {
+ SubmitOrchestrationRequest req = new SubmitOrchestrationRequest(
+ "idem-1", EscrowOperationType.CONFIRM_COMPLETION, "CABC", "4\"2", "AAAA==");
+ Set> violations = validator.validate(req);
+ assertThat(violations).anyMatch(v -> v.getPropertyPath().toString().equals("operationRef"));
+ }
+}
diff --git a/backend-api/src/test/java/com/guildworkman/api/escrow/rpc/SorobanRpcClientTest.java b/backend-api/src/test/java/com/guildworkman/api/escrow/rpc/SorobanRpcClientTest.java
new file mode 100644
index 0000000..3f8f61e
--- /dev/null
+++ b/backend-api/src/test/java/com/guildworkman/api/escrow/rpc/SorobanRpcClientTest.java
@@ -0,0 +1,123 @@
+package com.guildworkman.api.escrow.rpc;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.OkHttpClient;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SorobanRpcClientTest {
+
+ private MockWebServer server;
+ private SorobanRpcClient client;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ server = new MockWebServer();
+ server.start();
+
+ SorobanRpcProperties properties = new SorobanRpcProperties();
+ properties.setUrl(server.url("/").toString());
+
+ client = new SorobanRpcClient(new OkHttpClient(), new ObjectMapper(), properties);
+ }
+
+ @AfterEach
+ void tearDown() throws IOException {
+ server.shutdown();
+ }
+
+ @Test
+ void sendTransactionParsesAcceptedResult() throws InterruptedException {
+ server.enqueue(new MockResponse()
+ .setBody("{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"status\":\"PENDING\",\"hash\":\"abc123\"}}")
+ .addHeader("Content-Type", "application/json"));
+
+ SendTransactionResult result = client.sendTransaction("AAAA==");
+
+ assertThat(result.status()).isEqualTo("PENDING");
+ assertThat(result.hash()).isEqualTo("abc123");
+ assertThat(result.isAccepted()).isTrue();
+
+ RecordedRequest recorded = server.takeRequest();
+ assertThat(recorded.getBody().readUtf8()).contains("\"method\":\"sendTransaction\"").contains("AAAA==");
+ }
+
+ @Test
+ void getTransactionParsesSuccessResult() {
+ server.enqueue(new MockResponse()
+ .setBody("{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"status\":\"SUCCESS\",\"resultXdr\":\"xyz\",\"ledger\":100}}")
+ .addHeader("Content-Type", "application/json"));
+
+ GetTransactionResult result = client.getTransaction("hash123");
+
+ assertThat(result.isSuccess()).isTrue();
+ assertThat(result.ledger()).isEqualTo(100L);
+ }
+
+ @Test
+ void throwsOnJsonRpcErrorField() {
+ server.enqueue(new MockResponse()
+ .setBody("{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"error\":{\"code\":-32602,\"message\":\"invalid params\"}}")
+ .addHeader("Content-Type", "application/json"));
+
+ assertThatThrownBy(() -> client.sendTransaction("AAAA=="))
+ .isInstanceOf(SorobanRpcException.class)
+ .hasMessageContaining("invalid params");
+ }
+
+ @Test
+ void throwsOnHttpError() {
+ server.enqueue(new MockResponse().setResponseCode(500).setBody("boom"));
+
+ assertThatThrownBy(() -> client.getTransaction("hash123"))
+ .isInstanceOf(SorobanRpcException.class)
+ .hasMessageContaining("500");
+ }
+
+ // --- request-XDR log/exception safety -----------------------------------
+
+ @Test
+ void exceptionMessageNeverContainsTheFullSignedXdrOnHttpError() {
+ String largeXdr = "X".repeat(5000);
+ server.enqueue(new MockResponse().setResponseCode(500).setBody("unrelated server error, no echo"));
+
+ assertThatThrownBy(() -> client.sendTransaction(largeXdr))
+ .isInstanceOf(SorobanRpcException.class)
+ .hasMessageNotContaining(largeXdr);
+ }
+
+ @Test
+ void exceptionMessageNeverContainsTheFullSignedXdrOnJsonRpcError() {
+ String largeXdr = "Y".repeat(5000);
+ server.enqueue(new MockResponse()
+ .setBody("{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"error\":{\"code\":-32602,\"message\":\"invalid params\"}}")
+ .addHeader("Content-Type", "application/json"));
+
+ assertThatThrownBy(() -> client.sendTransaction(largeXdr))
+ .isInstanceOf(SorobanRpcException.class)
+ .hasMessageNotContaining(largeXdr);
+ }
+
+ @Test
+ void aServerResponseThatEchoesTheRequestIsTruncatedInTheExceptionMessage() {
+ String largeXdr = "Z".repeat(5000);
+ // Simulates a pathological RPC error response that echoes the
+ // offending request body back — the response is what gets truncated,
+ // not omitted, since it's still useful for debugging real RPC errors.
+ server.enqueue(new MockResponse().setResponseCode(400).setBody("bad request, you sent: " + largeXdr));
+
+ assertThatThrownBy(() -> client.sendTransaction(largeXdr))
+ .isInstanceOf(SorobanRpcException.class)
+ .hasMessageNotContaining(largeXdr)
+ .satisfies(ex -> assertThat(ex.getMessage().length()).isLessThan(largeXdr.length()));
+ }
+}
diff --git a/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowOrchestrationServiceTest.java b/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowOrchestrationServiceTest.java
new file mode 100644
index 0000000..366afce
--- /dev/null
+++ b/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowOrchestrationServiceTest.java
@@ -0,0 +1,267 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.escrow.api.SubmitOrchestrationRequest;
+import com.guildworkman.api.escrow.model.EscrowOperationType;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import com.guildworkman.api.escrow.rpc.GetTransactionResult;
+import com.guildworkman.api.escrow.rpc.SendTransactionResult;
+import com.guildworkman.api.escrow.rpc.SorobanRpcClient;
+import com.guildworkman.api.escrow.rpc.SorobanRpcException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DataIntegrityViolationException;
+
+import java.time.Instant;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+class EscrowOrchestrationServiceTest {
+
+ private EscrowOrchestrationRequestRepository repository;
+ private EscrowOrchestrationInserter inserter;
+ private SorobanRpcClient sorobanRpcClient;
+ private EscrowOrchestrationRetryProperties retryProperties;
+ private EscrowOrchestrationService service;
+
+ @BeforeEach
+ void setUp() {
+ repository = mock(EscrowOrchestrationRequestRepository.class);
+ inserter = mock(EscrowOrchestrationInserter.class);
+ sorobanRpcClient = mock(SorobanRpcClient.class);
+ retryProperties = new EscrowOrchestrationRetryProperties();
+ service = new EscrowOrchestrationService(repository, inserter, sorobanRpcClient, retryProperties);
+ }
+
+ private static SubmitOrchestrationRequest request(String key) {
+ return new SubmitOrchestrationRequest(key, EscrowOperationType.CONFIRM_COMPLETION, "CABC", "42", "AAAA==");
+ }
+
+ private static EscrowOrchestrationRequest entity(Long id, String key, OrchestrationStatus status) {
+ EscrowOrchestrationRequest e = new EscrowOrchestrationRequest();
+ e.setId(id);
+ e.setIdempotencyKey(key);
+ e.setOperationType(EscrowOperationType.CONFIRM_COMPLETION);
+ e.setContractId("CABC");
+ e.setOperationRef("42");
+ e.setSignedTransactionXdr("AAAA==");
+ e.setStatus(status);
+ e.setReconciliationStatus(ReconciliationStatus.PENDING);
+ return e;
+ }
+
+ // --- submit() idempotency ------------------------------------------------
+
+ @Test
+ void submitInsertsNewRequestForNewIdempotencyKey() {
+ var req = request("idem-1");
+ when(repository.findByIdempotencyKey("idem-1")).thenReturn(Optional.empty());
+ var saved = entity(1L, "idem-1", OrchestrationStatus.PENDING);
+ when(inserter.insert(req)).thenReturn(saved);
+
+ var result = service.submit(req);
+
+ assertThat(result.request().getId()).isEqualTo(1L);
+ assertThat(result.replayed()).isFalse();
+ verify(inserter).insert(req);
+ }
+
+ @Test
+ void submitReturnsExistingRequestOnDuplicateIdempotencyKey() {
+ var req = request("idem-dup");
+ var existing = entity(1L, "idem-dup", OrchestrationStatus.SUBMITTED);
+ when(repository.findByIdempotencyKey("idem-dup")).thenReturn(Optional.of(existing));
+
+ var result = service.submit(req);
+
+ assertThat(result.request().getId()).isEqualTo(1L);
+ assertThat(result.replayed()).isTrue();
+ verify(inserter, never()).insert(any());
+ }
+
+ @Test
+ void submitHandlesDataIntegrityViolationWithFallbackLookup() {
+ var req = request("race");
+ when(repository.findByIdempotencyKey("race")).thenReturn(Optional.empty());
+ when(inserter.insert(req)).thenThrow(new DataIntegrityViolationException("dup key"));
+ var winner = entity(9L, "race", OrchestrationStatus.PENDING);
+ when(repository.findByIdempotencyKey("race")).thenReturn(Optional.empty(), Optional.of(winner));
+
+ var result = service.submit(req);
+
+ assertThat(result.request().getId()).isEqualTo(9L);
+ assertThat(result.replayed()).isTrue();
+ }
+
+ @Test
+ void getThrowsWhenNotFound() {
+ when(repository.findById(404L)).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> service.get(404L))
+ .isInstanceOf(EscrowOrchestrationNotFoundException.class);
+ }
+
+ // --- requeueReconciliation() ---------------------------------------------
+
+ @Test
+ void requeueReconciliationResetsAMismatchedRequestToPending() {
+ var e = entity(1L, "k", OrchestrationStatus.CONFIRMED);
+ e.setReconciliationStatus(ReconciliationStatus.MISMATCHED);
+ e.setReconciledAt(Instant.now());
+ when(repository.findById(1L)).thenReturn(Optional.of(e));
+
+ var result = service.requeueReconciliation(1L);
+
+ assertThat(result.getReconciliationStatus()).isEqualTo(ReconciliationStatus.PENDING);
+ assertThat(result.getReconciledAt()).isNull();
+ verify(repository).save(e);
+ }
+
+ @Test
+ void requeueReconciliationRejectsANonMismatchedRequest() {
+ var e = entity(1L, "k", OrchestrationStatus.CONFIRMED);
+ e.setReconciliationStatus(ReconciliationStatus.MATCHED);
+ when(repository.findById(1L)).thenReturn(Optional.of(e));
+
+ assertThatThrownBy(() -> service.requeueReconciliation(1L))
+ .isInstanceOf(ReconciliationRequeueNotAllowedException.class);
+ verify(repository, never()).save(any());
+ }
+
+ @Test
+ void requeueReconciliationThrowsWhenNotFound() {
+ when(repository.findById(404L)).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> service.requeueReconciliation(404L))
+ .isInstanceOf(EscrowOrchestrationNotFoundException.class);
+ }
+
+ // --- submitOne() -----------------------------------------------------------
+
+ @Test
+ void submitOneTransitionsToSubmittedOnAcceptedResult() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ when(sorobanRpcClient.sendTransaction("AAAA==")).thenReturn(new SendTransactionResult("hash123", "PENDING", null));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.SUBMITTED);
+ assertThat(e.getSorobanTxHash()).isEqualTo("hash123");
+ assertThat(e.getSubmittedAt()).isNotNull();
+ verify(repository).save(e);
+ }
+
+ @Test
+ void submitOneTreatsDuplicateAsAccepted() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ when(sorobanRpcClient.sendTransaction(any())).thenReturn(new SendTransactionResult("hash123", "DUPLICATE", null));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.SUBMITTED);
+ }
+
+ @Test
+ void submitOneFailsTerminallyOnRpcErrorStatus() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ when(sorobanRpcClient.sendTransaction(any())).thenReturn(new SendTransactionResult(null, "ERROR", "bad-xdr"));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.FAILED);
+ assertThat(e.getLastError()).contains("bad-xdr");
+ }
+
+ @Test
+ void submitOneRetriesOnTryAgainLater() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ e.setAttempts(0);
+ when(sorobanRpcClient.sendTransaction(any())).thenReturn(new SendTransactionResult(null, "TRY_AGAIN_LATER", null));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.PENDING);
+ assertThat(e.getAttempts()).isEqualTo(1);
+ assertThat(e.getNextAttemptAt()).isAfter(Instant.now());
+ }
+
+ @Test
+ void submitOneMovesToDeadLetterAfterMaxAttempts() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ e.setAttempts(retryProperties.getMaxAttempts() - 1);
+ when(sorobanRpcClient.sendTransaction(any())).thenThrow(new SorobanRpcException("timeout"));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.DEAD_LETTER);
+ assertThat(e.getAttempts()).isEqualTo(retryProperties.getMaxAttempts());
+ }
+
+ @Test
+ void submitOneRetriesOnRpcException() {
+ var e = entity(1L, "k", OrchestrationStatus.PENDING);
+ e.setAttempts(0);
+ when(sorobanRpcClient.sendTransaction(any())).thenThrow(new SorobanRpcException("network error"));
+
+ service.submitOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.PENDING);
+ assertThat(e.getLastError()).isEqualTo("network error");
+ }
+
+ // --- pollOne() ---------------------------------------------------------
+
+ @Test
+ void pollOneConfirmsOnSuccess() {
+ var e = entity(1L, "k", OrchestrationStatus.SUBMITTED);
+ e.setSorobanTxHash("hash123");
+ when(sorobanRpcClient.getTransaction("hash123")).thenReturn(new GetTransactionResult("SUCCESS", "resultXdr", 100L));
+
+ service.pollOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.CONFIRMED);
+ assertThat(e.getConfirmedAt()).isNotNull();
+ }
+
+ @Test
+ void pollOneFailsTerminallyOnChainFailure() {
+ var e = entity(1L, "k", OrchestrationStatus.SUBMITTED);
+ e.setSorobanTxHash("hash123");
+ when(sorobanRpcClient.getTransaction("hash123")).thenReturn(new GetTransactionResult("FAILED", "resultXdr", 100L));
+
+ service.pollOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.FAILED);
+ }
+
+ @Test
+ void pollOneKeepsPollingOnNotFound() {
+ var e = entity(1L, "k", OrchestrationStatus.SUBMITTED);
+ e.setSorobanTxHash("hash123");
+ e.setAttempts(0);
+ when(sorobanRpcClient.getTransaction("hash123")).thenReturn(new GetTransactionResult("NOT_FOUND", null, null));
+
+ service.pollOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.SUBMITTED);
+ assertThat(e.getAttempts()).isEqualTo(1);
+ }
+
+ @Test
+ void pollOneMovesToDeadLetterAfterMaxAttemptsNotFound() {
+ var e = entity(1L, "k", OrchestrationStatus.SUBMITTED);
+ e.setSorobanTxHash("hash123");
+ e.setAttempts(retryProperties.getMaxAttempts() - 1);
+ when(sorobanRpcClient.getTransaction("hash123")).thenReturn(new GetTransactionResult("NOT_FOUND", null, null));
+
+ service.pollOne(e);
+
+ assertThat(e.getStatus()).isEqualTo(OrchestrationStatus.DEAD_LETTER);
+ }
+}
diff --git a/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowReconciliationServiceTest.java b/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowReconciliationServiceTest.java
new file mode 100644
index 0000000..25222d0
--- /dev/null
+++ b/backend-api/src/test/java/com/guildworkman/api/escrow/service/EscrowReconciliationServiceTest.java
@@ -0,0 +1,105 @@
+package com.guildworkman.api.escrow.service;
+
+import com.guildworkman.api.chain.model.ChainEventStatus;
+import com.guildworkman.api.chain.model.OnChainEvent;
+import com.guildworkman.api.chain.repository.OnChainEventRepository;
+import com.guildworkman.api.escrow.model.EscrowOperationType;
+import com.guildworkman.api.escrow.model.EscrowOrchestrationRequest;
+import com.guildworkman.api.escrow.model.OrchestrationStatus;
+import com.guildworkman.api.escrow.model.ReconciliationStatus;
+import com.guildworkman.api.escrow.repository.EscrowOrchestrationRequestRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class EscrowReconciliationServiceTest {
+
+ private EscrowOrchestrationRequestRepository orchestrationRequests;
+ private OnChainEventRepository onChainEvents;
+ private EscrowReconciliationProperties properties;
+ private EscrowReconciliationService service;
+
+ @BeforeEach
+ void setUp() {
+ orchestrationRequests = mock(EscrowOrchestrationRequestRepository.class);
+ onChainEvents = mock(OnChainEventRepository.class);
+ properties = new EscrowReconciliationProperties();
+ properties.setWindow(Duration.ofMinutes(10));
+ service = new EscrowReconciliationService(orchestrationRequests, onChainEvents, properties);
+ }
+
+ private static EscrowOrchestrationRequest confirmed(String contractId, String operationRef, Instant confirmedAt) {
+ EscrowOrchestrationRequest e = new EscrowOrchestrationRequest();
+ e.setId(1L);
+ e.setContractId(contractId);
+ e.setOperationRef(operationRef);
+ e.setOperationType(EscrowOperationType.CONFIRM_COMPLETION);
+ e.setStatus(OrchestrationStatus.CONFIRMED);
+ e.setReconciliationStatus(ReconciliationStatus.PENDING);
+ e.setConfirmedAt(confirmedAt);
+ return e;
+ }
+
+ private static OnChainEvent event(ChainEventStatus status) {
+ OnChainEvent e = new OnChainEvent();
+ e.setId(1L);
+ e.setStatus(status);
+ return e;
+ }
+
+ @Test
+ void marksMatchedWhenProcessedOnChainEventFound() {
+ var request = confirmed("CABC", "42", Instant.now());
+ when(onChainEvents.findByContractIdAndTopicsContaining(eq("CABC"), eq("\"42\"")))
+ .thenReturn(List.of(event(ChainEventStatus.PROCESSED)));
+
+ service.reconcileOne(request);
+
+ assertThat(request.getReconciliationStatus()).isEqualTo(ReconciliationStatus.MATCHED);
+ assertThat(request.getReconciledAt()).isNotNull();
+ verify(orchestrationRequests).save(request);
+ }
+
+ @Test
+ void staysPendingWhenMatchingEventNotYetProcessed() {
+ var request = confirmed("CABC", "42", Instant.now());
+ when(onChainEvents.findByContractIdAndTopicsContaining(any(), any()))
+ .thenReturn(List.of(event(ChainEventStatus.PENDING)));
+
+ service.reconcileOne(request);
+
+ assertThat(request.getReconciliationStatus()).isEqualTo(ReconciliationStatus.PENDING);
+ }
+
+ @Test
+ void staysPendingWithinGraceWindowWhenNoEventFound() {
+ var request = confirmed("CABC", "42", Instant.now());
+ when(onChainEvents.findByContractIdAndTopicsContaining(any(), any())).thenReturn(List.of());
+
+ service.reconcileOne(request);
+
+ assertThat(request.getReconciliationStatus()).isEqualTo(ReconciliationStatus.PENDING);
+ }
+
+ @Test
+ void marksMismatchedAfterGraceWindowElapsesWithNoEvent() {
+ var request = confirmed("CABC", "42", Instant.now().minus(Duration.ofMinutes(11)));
+ when(onChainEvents.findByContractIdAndTopicsContaining(any(), any())).thenReturn(List.of());
+
+ service.reconcileOne(request);
+
+ assertThat(request.getReconciliationStatus()).isEqualTo(ReconciliationStatus.MISMATCHED);
+ assertThat(request.getReconciledAt()).isNotNull();
+ verify(orchestrationRequests).save(request);
+ }
+}