[#1941] Add endpoint broadcaster API

CHANGELOG.md

@@ -7,6 +7,10 @@ and this library adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- `Synchronizer.broadcaster` API for creating transactions without immediate
+  submission and submitting stored transactions to selected lightwalletd
+  endpoints. Automatic retry uses the endpoints submitted through the
+  broadcaster.
 - `Synchronizer.fullyScannedHeight` and `Synchronizer.getTreeState` accessors
   for snapshot-height consumers.
 

sdk-lib/build.gradle.kts

@@ -135,6 +135,7 @@ dependencies {
     testImplementation(libs.kotlin.reflect)
     testImplementation(libs.kotlin.test)
     testImplementation(libs.bundles.junit)
+    testImplementation(libs.mockito.junit)
 
     // NOTE: androidTests will use JUnit4, while src/test/java tests will leverage Junit5
     // Attempting to use JUnit5 via https://github.com/mannodermaus/android-junit5 was painful. The plugin configuration

sdk-lib/src/main/java/cash/z/ecc/android/sdk/Broadcaster.kt

@@ -0,0 +1,50 @@
+package cash.z.ecc.android.sdk
+
+import cash.z.ecc.android.sdk.model.CreatedTransaction
+import cash.z.ecc.android.sdk.model.Pczt
+import cash.z.ecc.android.sdk.model.Proposal
+import cash.z.ecc.android.sdk.model.TransactionSubmitResult
+import cash.z.ecc.android.sdk.model.UnifiedSpendingKey
+import co.electriccoin.lightwallet.client.model.LightWalletEndpoint
+
+/**
+ * Creates transactions without immediately submitting them, and submits stored
+ * transaction bytes to a specific lightwalletd endpoint.
+ *
+ * Transactions created through this API wait for the caller to submit them.
+ * Once submitted, automatic retry uses the submitted endpoints instead of the
+ * synchronizer's default endpoint.
+ */
+interface Broadcaster {
+    /**
+     * Creates and stores the transactions in [proposal] without submitting them.
+     *
+     * Created transactions will not be automatically resubmitted until they are
+     * submitted through this API.
+     */
+    suspend fun createProposedTransactions(
+        proposal: Proposal,
+        usk: UnifiedSpendingKey
+    ): List<CreatedTransaction>
+
+    /**
+     * Finalizes and stores a separately proven and signed PCZT without submitting it.
+     *
+     * Created transactions will not be automatically resubmitted until they are
+     * submitted through this API.
+     */
+    suspend fun createTransactionFromPczt(
+        pcztWithProofs: Pczt,
+        pcztWithSignatures: Pczt
+    ): List<CreatedTransaction>
+
+    /**
+     * Submits [transaction] to the provided [endpoint].
+     *
+     * The endpoint is also remembered for automatic retry.
+     */
+    suspend fun submit(
+        transaction: CreatedTransaction,
+        endpoint: LightWalletEndpoint
+    ): TransactionSubmitResult
+}

sdk-lib/src/main/java/cash/z/ecc/android/sdk/SdkSynchronizer.kt

@@ -46,8 +46,12 @@ import cash.z.ecc.android.sdk.internal.storage.preference.EncryptedPreferencePro
 import cash.z.ecc.android.sdk.internal.storage.preference.StandardPreferenceProvider
 import cash.z.ecc.android.sdk.internal.storage.preference.api.PreferenceProvider
 import cash.z.ecc.android.sdk.internal.storage.preference.keys.StandardPreferenceKeys.SDK_VERSION_OF_LAST_FIX_WITNESSES_CALL
+import cash.z.ecc.android.sdk.internal.transaction.EndpointTransactionSubmitter
 import cash.z.ecc.android.sdk.internal.transaction.OutboundTransactionManager
 import cash.z.ecc.android.sdk.internal.transaction.OutboundTransactionManagerImpl
+import cash.z.ecc.android.sdk.internal.transaction.PendingSubmitPlanStore
+import cash.z.ecc.android.sdk.internal.transaction.SdkBroadcaster
+import cash.z.ecc.android.sdk.internal.transaction.SubmitPlanExecutor
 import cash.z.ecc.android.sdk.internal.transaction.TransactionEncoder
 import cash.z.ecc.android.sdk.internal.transaction.TransactionEncoderImpl
 import cash.z.ecc.android.sdk.model.Account
@@ -105,14 +109,12 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
@@ -157,6 +159,8 @@ class SdkSynchronizer private constructor(
     private val torClient: TorClient?,
     private val walletClient: CombinedWalletClient,
     private val walletClientFactory: WalletClientFactory,
+    private val defaultSubmitEndpoint: LightWalletEndpoint,
+    private val pendingSubmitPlanStore: PendingSubmitPlanStore,
     private val sdkFlags: SdkFlags
 ) : CloseableSynchronizer {
     companion object {
@@ -197,6 +201,8 @@ class SdkSynchronizer private constructor(
             torClient: TorClient?,
             walletClient: CombinedWalletClient,
             walletClientFactory: WalletClientFactory,
+            defaultSubmitEndpoint: LightWalletEndpoint,
+            pendingSubmitPlanStore: PendingSubmitPlanStore,
             sdkFlags: SdkFlags
         ): CloseableSynchronizer {
             val synchronizerKey = SynchronizerKey(zcashNetwork, alias)
@@ -216,6 +222,8 @@ class SdkSynchronizer private constructor(
                     torClient = torClient,
                     walletClient = walletClient,
                     walletClientFactory = walletClientFactory,
+                    defaultSubmitEndpoint = defaultSubmitEndpoint,
+                    pendingSubmitPlanStore = pendingSubmitPlanStore,
                     sdkFlags = sdkFlags
                 ).apply {
                     instances[synchronizerKey] = InstanceState.Active
@@ -289,6 +297,19 @@ class SdkSynchronizer private constructor(
 
     val coroutineScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Main)
 
+    private val sdkBroadcaster =
+        SdkBroadcaster(
+            txManager = txManager,
+            transactionSubmitter =
+                EndpointTransactionSubmitter(
+                    walletClientFactory = walletClientFactory,
+                    sdkFlags = sdkFlags
+                ),
+            pendingSubmitPlanStore = pendingSubmitPlanStore
+        )
+
+    override val broadcaster: Broadcaster = sdkBroadcaster
+
     override val walletBalances = processor.walletBalances.asStateFlow()
 
     private val refreshExchangeRateUsd = MutableSharedFlow<Unit>()
@@ -1061,30 +1082,13 @@ class SdkSynchronizer private constructor(
         proposal: Proposal,
         usk: UnifiedSpendingKey
     ): Flow<TransactionSubmitResult> {
-        // Internally, this logic submits and checks every incoming transaction, and once [Failure] or
-        // [NotAttempted] submission result occurs, it returns [NotAttempted] for the rest of them
-        var anySubmissionFailed = false
-        return txManager
-            .createProposedTransactions(proposal, usk)
-            .asFlow()
-            .map { transaction ->
-                if (anySubmissionFailed) {
-                    TransactionSubmitResult.NotAttempted(transaction.txId)
-                } else {
-                    val submission = txManager.submit(transaction)
-                    when (submission) {
-                        is TransactionSubmitResult.Success -> {
-                            // Expected state
-                        }
-
-                        is TransactionSubmitResult.Failure,
-                        is TransactionSubmitResult.NotAttempted -> {
-                            anySubmissionFailed = true
-                        }
-                    }
-                    submission
-                }
-            }
+        // This preserves the legacy API contract by creating locally, then submitting each
+        // created transaction to the builder-configured default endpoint.
+        return sdkBroadcaster.createAndSubmitProposedTransactions(
+            proposal = proposal,
+            usk = usk,
+            endpoint = defaultSubmitEndpoint
+        )
     }
 
     override suspend fun createPcztFromProposal(
@@ -1102,9 +1106,13 @@ class SdkSynchronizer private constructor(
         pcztWithProofs: Pczt,
         pcztWithSignatures: Pczt
     ): Flow<TransactionSubmitResult> {
-        // Internally, this logic submits and checks the newly stored and encoded transaction
-        return flowOf(txManager.extractAndStoreTxFromPczt(pcztWithProofs, pcztWithSignatures))
-            .map { transaction -> txManager.submit(transaction) }
+        // This preserves the legacy API contract by submitting the stored transaction to the
+        // builder-configured default endpoint.
+        return sdkBroadcaster.createAndSubmitTransactionFromPczt(
+            pcztWithProofs = pcztWithProofs,
+            pcztWithSignatures = pcztWithSignatures,
+            endpoint = defaultSubmitEndpoint
+        )
     }
 
     override suspend fun refreshUtxos(
@@ -1386,7 +1394,9 @@ internal object DefaultSynchronizerFactory {
         birthdayHeight: BlockHeight,
         txManager: OutboundTransactionManager,
         sdkFlags: SdkFlags,
-        saplingParamFetcher: SaplingParamFetcher
+        saplingParamFetcher: SaplingParamFetcher,
+        pendingSubmitPlanStore: PendingSubmitPlanStore,
+        submitPlanExecutor: SubmitPlanExecutor
     ): CompactBlockProcessor =
         CompactBlockProcessor(
             backend = backend,
@@ -1395,7 +1405,9 @@ internal object DefaultSynchronizerFactory {
             repository = repository,
             txManager = txManager,
             sdkFlags = sdkFlags,
-            saplingParamFetcher = saplingParamFetcher
+            saplingParamFetcher = saplingParamFetcher,
+            pendingSubmitPlanStore = pendingSubmitPlanStore,
+            submitPlanExecutor = submitPlanExecutor
         )
 }
 

sdk-lib/src/main/java/cash/z/ecc/android/sdk/Synchronizer.kt

@@ -19,13 +19,18 @@ import cash.z.ecc.android.sdk.internal.db.DatabaseCoordinator
 import cash.z.ecc.android.sdk.internal.exchange.UsdExchangeRateFetcher
 import cash.z.ecc.android.sdk.internal.model.TorClient
 import cash.z.ecc.android.sdk.internal.model.ext.toBlockHeight
+import cash.z.ecc.android.sdk.internal.storage.preference.EncryptedPreferenceProvider
 import cash.z.ecc.android.sdk.internal.storage.preference.StandardPreferenceProvider
+import cash.z.ecc.android.sdk.internal.transaction.EndpointTransactionSubmitter
+import cash.z.ecc.android.sdk.internal.transaction.PendingSubmitPlanStore
+import cash.z.ecc.android.sdk.internal.transaction.SubmitPlanExecutor
 import cash.z.ecc.android.sdk.model.Account
 import cash.z.ecc.android.sdk.model.AccountBalance
 import cash.z.ecc.android.sdk.model.AccountCreateSetup
 import cash.z.ecc.android.sdk.model.AccountImportSetup
 import cash.z.ecc.android.sdk.model.AccountUuid
 import cash.z.ecc.android.sdk.model.BlockHeight
+import cash.z.ecc.android.sdk.model.CreatedTransaction
 import cash.z.ecc.android.sdk.model.FastestServersResult
 import cash.z.ecc.android.sdk.model.ObserveFiatCurrencyResult
 import cash.z.ecc.android.sdk.model.Pczt
@@ -142,6 +147,16 @@ interface Synchronizer {
      */
     val latestBirthdayHeight: BlockHeight?
 
+    /**
+     * Creates transactions without submitting them and submits created transactions
+     * to caller-selected lightwalletd endpoints.
+     *
+     * The default implementation throws [UnsupportedOperationException]. SDK-backed
+     * synchronizers override this with a working broadcaster.
+     */
+    val broadcaster: Broadcaster
+        get() = UnavailableBroadcaster
+
     //
     // Operations
     //
@@ -949,6 +964,20 @@ interface Synchronizer {
             val encoder = DefaultSynchronizerFactory.defaultEncoder(backend, saplingParamFetcher, repository)
 
             val txManager = DefaultSynchronizerFactory.defaultTxManager(encoder, walletClient, sdkFlags)
+            val standardPreferenceProvider = StandardPreferenceProvider(context)
+            val preferenceProvider = standardPreferenceProvider()
+            val encryptedPreferenceProvider = EncryptedPreferenceProvider(applicationContext)
+            val pendingSubmitPlanStore =
+                PendingSubmitPlanStore(
+                    preferenceProvider = encryptedPreferenceProvider(),
+                    namespace = "${zcashNetwork.id}_$alias"
+                )
+            val transactionSubmitter =
+                EndpointTransactionSubmitter(
+                    walletClientFactory = walletClientFactory,
+                    sdkFlags = sdkFlags
+                )
+            val submitPlanExecutor = SubmitPlanExecutor(transactionSubmitter)
             val processor =
                 DefaultSynchronizerFactory.defaultProcessor(
                     backend = backend,
@@ -957,11 +986,11 @@ interface Synchronizer {
                     repository = repository,
                     txManager = txManager,
                     sdkFlags = sdkFlags,
-                    saplingParamFetcher = saplingParamFetcher
+                    saplingParamFetcher = saplingParamFetcher,
+                    pendingSubmitPlanStore = pendingSubmitPlanStore,
+                    submitPlanExecutor = submitPlanExecutor
                 )
 
-            val standardPreferenceProvider = StandardPreferenceProvider(context)
-
             return SdkSynchronizer.new(
                 context = context.applicationContext,
                 zcashNetwork = zcashNetwork,
@@ -979,10 +1008,12 @@ interface Synchronizer {
                     ),
                 fetchExchangeChangeUsd =
                     exchangeRateIsolatedTorClient?.let { UsdExchangeRateFetcher(isolatedTorClient = it) },
-                preferenceProvider = standardPreferenceProvider(),
+                preferenceProvider = preferenceProvider,
                 torClient = torClient,
                 walletClient = walletClient,
                 walletClientFactory = walletClientFactory,
+                defaultSubmitEndpoint = lightWalletEndpoint,
+                pendingSubmitPlanStore = pendingSubmitPlanStore,
                 sdkFlags = sdkFlags
             )
         }
@@ -1042,6 +1073,29 @@ interface Synchronizer {
     }
 }
 
+private object UnavailableBroadcaster : Broadcaster {
+    override suspend fun createProposedTransactions(
+        proposal: Proposal,
+        usk: UnifiedSpendingKey
+    ): List<CreatedTransaction> = throw UnsupportedOperationException(
+        "Synchronizer.broadcaster is unavailable for this Synchronizer implementation."
+    )
+
+    override suspend fun createTransactionFromPczt(
+        pcztWithProofs: Pczt,
+        pcztWithSignatures: Pczt
+    ): List<CreatedTransaction> = throw UnsupportedOperationException(
+        "Synchronizer.broadcaster is unavailable for this Synchronizer implementation."
+    )
+
+    override suspend fun submit(
+        transaction: CreatedTransaction,
+        endpoint: LightWalletEndpoint
+    ): TransactionSubmitResult = throw UnsupportedOperationException(
+        "Synchronizer.broadcaster is unavailable for this Synchronizer implementation."
+    )
+}
+
 /**
  * Sealed class describing wallet initialization mode.
  *