fix(utxo): pick in-batch duplicate spend winners from persisted state

CHANGELOG.md

@@ -327,6 +327,54 @@ Upgrading from v0.16.0 changes nothing for them; upgrading from an earlier relea
   (including a non-positive `subtreevalidation_txMetaTombstoneWindow`). It used to
   log the failure and run with no store, then panic on first use.
 
+- **UTXO batch conflicts are now decided by the stored owner.** SQL and Aerospike
+  send competing claims on the same outpoint to the backend in ordered parts. A
+  re-submitted spend is accepted when it matches the stored owner, even if an
+  earlier claim in the batch fails. Double-spend errors name the actual owner,
+  including owners absent from the batch. Previously, arrival order could cause a
+  false `UTXO_SPENT` error and rollback of an existing spend when the spending
+  transaction had not yet been created. Callers can now see success instead of
+  `UTXO_SPENT` for valid retries. No configuration change is required.
+
+  Once a part tells the stored owner of an outpoint, later competing claims in
+  the batch are refused locally with `UTXO_SPENT` naming that owner, when they
+  run the same store checks as the claim that showed the owner: the same UTXO
+  hash, block height and ignore flags. Thus alternating double-spends cost two
+  backend calls, not one per claim. Non-`UTXO_SPENT` backend failures are reused
+  for later same-outpoint claims with identical checks, so a storage outage or
+  missing parent no longer causes a serial backend call per competing claim.
+  Claims with different checks still need their own decision; there is no global
+  two-part limit across independent outpoints or validation settings.
+
+  SQL and Aerospike now check each queued input's admission state and caller
+  deadline immediately before dispatching its backend part. A caller that stops
+  waiting abandons queued inputs before rollback, preventing later parts from
+  writing those inputs after the caller has returned. `utxostore_spendWaitTimeout`
+  now includes enqueue time (per input on SQL, per transaction on Aerospike), so
+  callers under queue pressure may receive timeout errors sooner. The setting
+  and its default are unchanged; no configuration change is required. Operations
+  already admitted before timeout can still finish late: this does not close the
+  general in-flight rollback window tracked in #1291.
+
+  The `teranode_aerospike_utxo_spend_batch` and
+  `teranode_aerospike_utxo_spend_batch_size` histograms now count one batcher
+  batch, however many parts it needs. The Lua retry of `FILTERED_OUT` records is
+  no longer a separate histogram sample. The Aerospike spend circuit breaker
+  records expression outcomes before Lua retries, including in split batches.
+  Repeated expression failures count once per batcher batch until a success
+  resets the breaker; the next failure then counts again. Repeated expression
+  successes are likewise coalesced until a failure. Lua records every success
+  and infrastructure failure in processing order, including retries and repeated
+  keys in later parts. Split Lua batches can therefore reach the breaker
+  threshold sooner during an outage because they perform more backend operations.
+  Breaker settings are unchanged. The SQL spend batcher now
+  recovers a panic and answers every waiting caller with an error, instead of
+  leaving them to wait for `utxostore_spendWaitTimeout`. Go integrations using
+  `FilterConflictingDuplicateSpendClaims` must migrate to `ProcessSpendBatch`
+  with a `SpendBatchHandler`; omitting `SameChecks` disables local result reuse.
+  Asynchronous integrations should supply
+  `Start` to exclude abandoned requests at each part boundary. (#2036, bitcoin-sv/teranode#4934)
+
 - **A failed revert of an invalid block's mined tags now holds block assembly until it
   succeeds.** When block validation reverts the mined tags of an invalid block and
   `SetMinedMulti` fails, or a subtree read fails with anything other than not-found, the
@@ -892,21 +940,21 @@ Upgrading from v0.16.0 changes nothing for them; upgrading from an earlier relea
 - **Conflict-check failures in subtree validation are no longer reported as
   context cancellations.** Two follow-ups to the entry above, both in subtree
   validation's conflict check:
-  - On the fresh path (`CheckBlockSubtrees` on a new subtree, `CheckSubtreeFromBlock`,
-    live peer subtrees), a store fault, the `utxostore_conflictingChildrenMaxNodes`
-    walk limit, a frozen output, an input out of range and the fail-fast
-    missing-tx threshold are no longer wrapped as a context cancellation. Each is
-    still retried and never charged to a peer. What changes is how they show up:
-    on the live path `subtreeMessageHandler` logs them at error level ("error
-    processing kafka message") where it logged a "Context canceled, skipping"
-    warning, and a catch-up run that ends on the walk limit, a frozen output or an
-    out-of-range input records `local_processing_fault` instead of
-    `local_context_cancelled` as its last error type. A missing record, a
-    cancelled check, and a record the store could not decode keep the old
-    classification.
-  - On the fresh path a proven conflict is rejected with `TX_INVALID` even when
-    the check's context is cancelled while it runs, as the cached path and
-    blessing a missing transaction already did. Before, it was retried.
+    - On the fresh path (`CheckBlockSubtrees` on a new subtree, `CheckSubtreeFromBlock`,
+      live peer subtrees), a store fault, the `utxostore_conflictingChildrenMaxNodes`
+      walk limit, a frozen output, an input out of range and the fail-fast
+      missing-tx threshold are no longer wrapped as a context cancellation. Each is
+      still retried and never charged to a peer. What changes is how they show up:
+      on the live path `subtreeMessageHandler` logs them at error level ("error
+      processing kafka message") where it logged a "Context canceled, skipping"
+      warning, and a catch-up run that ends on the walk limit, a frozen output or an
+      out-of-range input records `local_processing_fault` instead of
+      `local_context_cancelled` as its last error type. A missing record, a
+      cancelled check, and a record the store could not decode keep the old
+      classification.
+    - On the fresh path a proven conflict is rejected with `TX_INVALID` even when
+      the check's context is cancelled while it runs, as the cached path and
+      blessing a missing transaction already did. Before, it was retried.
 
   Nothing to configure. (bsv-blockchain/teranode#2066)
 

stores/utxo/aerospike/circuit_breaker.go

@@ -270,3 +270,38 @@ func (cb *circuitBreaker) reset() {
 	cb.halfOpenAttempts = 0
 	cb.consecutiveSuccess = 0
 }
+
+// spendBreakerOutcome coalesces repeated expression outcomes across parts until
+// an opposite outcome is recorded. An uninterrupted run of expression timeouts
+// counts once per batcher batch, but a success re-arms failure recording so a
+// later timeout cannot disappear after the breaker was reset. Outcomes are
+// recorded before Lua retries, preserving their processing order.
+// Lua records always count, even when a later part reuses the same record key.
+type spendBreakerOutcome struct {
+	success bool
+	failure bool
+}
+
+func (o *spendBreakerOutcome) recordExpressions(cb *circuitBreaker, success, failure bool) {
+	if success && !o.success {
+		o.record(cb, true)
+		o.success = true
+	}
+
+	if failure && !o.failure {
+		o.record(cb, false)
+		o.failure = true
+	}
+}
+
+// record also re-arms the opposite expression outcome: a Lua success resets
+// the breaker, so a subsequent expression failure must count again (and vice versa).
+func (o *spendBreakerOutcome) record(cb *circuitBreaker, success bool) {
+	if success {
+		o.failure = false
+		cb.RecordSuccess()
+	} else {
+		o.success = false
+		cb.RecordFailure()
+	}
+}

stores/utxo/aerospike/respend_batch_4934_test.go

@@ -0,0 +1,115 @@
+package aerospike_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/bsv-blockchain/go-bt/v2"
+	"github.com/bsv-blockchain/go-bt/v2/bscript"
+	"github.com/bsv-blockchain/go-bt/v2/chainhash"
+	"github.com/bsv-blockchain/teranode/errors"
+	"github.com/bsv-blockchain/teranode/stores/utxo"
+	"github.com/bsv-blockchain/teranode/ulogger"
+	"github.com/bsv-blockchain/teranode/util/test"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRespendBehindLoserInSameBatch is the Aerospike half of bitcoin-sv/teranode#4934.
+//
+// S0 already owns both parent outputs. A batch then carries a double-spend
+// attempt on output 0 AHEAD of S0's idempotent re-spend. The filter used to
+// award the outpoint by batch position, so the legitimate re-spend was refused
+// with ErrSpent naming the double-spend attempt, which itself was refused by the
+// backend as already spent -- both claims lost, each blamed on the other.
+//
+// A test-only adapter supplies an exact batch to both backend paths.
+func TestRespendBehindLoserInSameBatch(t *testing.T) {
+	cases := []struct {
+		name              string
+		utxoBatchSize     int
+		enableExpressions bool
+	}{
+		{name: "lua", utxoBatchSize: 128, enableExpressions: false},
+		{name: "expressions", utxoBatchSize: 1, enableExpressions: true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			tSettings := test.CreateBaseTestSettings(t)
+			tSettings.UtxoStore.UtxoBatchSize = tc.utxoBatchSize
+			tSettings.UtxoStore.SpendBatcherSize = 1
+			tSettings.UtxoStore.SpendBatcherDurationMillis = 1
+			tSettings.UtxoStore.SpendWaitTimeout = 30 * time.Second
+			tSettings.Aerospike.EnableSpendFilterExpressions = tc.enableExpressions
+
+			client, store, storeCtx, deferFn := initAerospike(t, tSettings, ulogger.NewErrorTestLogger(t))
+			t.Cleanup(deferFn)
+			cleanDB(t, client)
+			require.NoError(t, store.SetBlockHeight(2))
+
+			parent := bt.NewTx()
+			prev := chainhash.DoubleHashH([]byte("4934-as-" + tc.name))
+			require.NoError(t, parent.From(prev.String(), 0,
+				"76a91471d7dd96d9edda09180fe9d57a477b5acc9cad1188ac", 1e8))
+			parent.Inputs[0].UnlockingScript = bscript.NewFromBytes([]byte{bscript.OpTRUE})
+			require.NoError(t, parent.PayToAddress("1BUBHgpCBWCsD2Ajbeu7ZgCgvMHsoYHm3W", 1000))
+			require.NoError(t, parent.PayToAddress("1BUBHgpCBWCsD2Ajbeu7ZgCgvMHsoYHm3W", 2000))
+			_, err := store.Create(storeCtx, parent, 1)
+			require.NoError(t, err)
+
+			spendOf := func(lockTime uint32, vouts ...uint32) *bt.Tx {
+				tx := bt.NewTx()
+				for _, v := range vouts {
+					require.NoError(t, tx.From(parent.TxIDChainHash().String(), v,
+						parent.Outputs[v].LockingScript.String(), parent.Outputs[v].Satoshis))
+				}
+				require.NoError(t, tx.PayToAddress("1BUBHgpCBWCsD2Ajbeu7ZgCgvMHsoYHm3W", 1))
+				tx.LockTime = lockTime
+
+				return tx
+			}
+
+			respend := spendOf(1, 0, 1) // legitimate spender of both outputs
+			loser := spendOf(2, 0)      // double-spend attempt on output 0
+
+			_, err = store.Spend(storeCtx, respend, 2)
+			require.NoError(t, err, "initial spend must succeed")
+
+			spends, err := utxo.GetSpends(respend)
+			require.NoError(t, err)
+			require.Len(t, spends, 2)
+			loserSpends, err := utxo.GetSpends(loser)
+			require.NoError(t, err)
+			store.SendSpendBatchForTest(t, []*utxo.Spend{loserSpends[0], spends[0], spends[1]}, 2)
+			require.NoError(t, spends[0].Err, "the persisted owner must be accepted")
+			require.NoError(t, spends[1].Err)
+			require.ErrorIs(t, loserSpends[0].Err, errors.ErrSpent)
+			require.Contains(t, loserSpends[0].Err.Error(), respend.TxID())
+
+			for _, sp := range spends {
+				resp, getErr := store.GetSpend(storeCtx, sp)
+				require.NoError(t, getErr)
+				t.Logf("output %d: status=%v spendingData=%v", sp.Vout, resp.Status, resp.SpendingData)
+
+				require.Equal(t, int(utxo.Status_SPENT), resp.Status,
+					"output %d that S0 legitimately spent went back to unspent", sp.Vout)
+				require.NotNil(t, resp.SpendingData, "output %d must name its spender", sp.Vout)
+				require.Equal(t, respend.TxID(), resp.SpendingData.TxID.String(),
+					"output %d is owned by %s, not by S0", sp.Vout, resp.SpendingData.TxID)
+			}
+			// Neither claim matches the stored owner. Both errors must name it.
+			first, err := utxo.GetSpends(spendOf(3, 0))
+			require.NoError(t, err)
+			second, err := utxo.GetSpends(spendOf(4, 0))
+			require.NoError(t, err)
+			store.SendSpendBatchForTest(t, []*utxo.Spend{first[0], second[0]}, 2)
+			for _, sp := range []*utxo.Spend{first[0], second[0]} {
+				require.ErrorIs(t, sp.Err, errors.ErrSpent)
+				var data *errors.UtxoSpentErrData
+				require.True(t, errors.AsData(sp.Err, &data))
+				require.Equal(t, respend.TxID(), data.SpendingData.TxID.String())
+			}
+
+		})
+	}
+}