Fix identical outputs issue [#3422]

Closes #3422. Fixes an issue where tokens can be lost if a SendTokens or IssueTokens transaction has identical outputs (same recipient, same amount).

Changelog: fix

internal/core/execute/v2/block/block.go

@@ -21,6 +21,8 @@ type Block struct {
 	State    BlockState
 	Batch    *database.Batch
 	Executor *Executor
+
+	syntheticCount uint64
 }
 
 func (b *Block) Params() execute.BlockParams { return b.BlockParams }

internal/core/execute/v2/block/exec_process.go

@@ -14,7 +14,9 @@ import (
 	"gitlab.com/accumulatenetwork/accumulate/internal/database"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/errors"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/types/messaging"
+	"gitlab.com/accumulatenetwork/accumulate/pkg/url"
 	"gitlab.com/accumulatenetwork/accumulate/protocol"
+	"golang.org/x/exp/slog"
 )
 
 // bundle is a bundle of messages to be processed.
@@ -194,6 +196,20 @@ func (d *bundle) process() ([]*protocol.TransactionStatus, error) {
 		d.produced = append(d.produced, ctx.produced...)
 	}
 
+	// Check for duplicates. This is a serious error if it occurs, but returning
+	// an error here would effectively stall the network.
+	pCount := map[[32]byte]int{}
+	pID := map[[32]byte]*url.TxID{}
+	for _, m := range d.produced {
+		pCount[m.Message.Hash()]++
+		pID[m.Message.Hash()] = m.Message.ID()
+	}
+	for h, c := range pCount {
+		if c > 1 {
+			slog.ErrorCtx(d.Context, "Duplicate synthetic messages", "id", pID[h], "count", c)
+		}
+	}
+
 	// Execute produced messages immediately if and only if the producer and
 	// destination are in the same domain. This implementation is inefficient
 	// but it preserves order and its good enough for now.

internal/core/execute/v2/block/msg_common.go

@@ -14,6 +14,7 @@ import (
 	"gitlab.com/accumulatenetwork/accumulate/pkg/types/messaging"
 	"gitlab.com/accumulatenetwork/accumulate/pkg/url"
 	"gitlab.com/accumulatenetwork/accumulate/protocol"
+	"golang.org/x/exp/slog"
 )
 
 // MessageContext is the context in which a message is processed.
@@ -151,14 +152,37 @@ func (m *MessageContext) queueAdditional(msg messaging.Message) {
 
 // didProduce queues a produced synthetic message for dispatch.
 func (m *MessageContext) didProduce(batch *database.Batch, dest *url.URL, msg messaging.Message) error {
-	if dest == nil {
-		panic("nil destination for produced message")
-	}
-	m.produced = append(m.produced, &ProducedMessage{
+	p := &ProducedMessage{
 		Producer:    m.message.ID(),
 		Destination: dest,
 		Message:     msg,
-	})
+	}
+
+	// Add an index to synthetic transactions to differentiate otherwise
+	// identical messages
+	m.syntheticCount++
+	switch msg := msg.(type) {
+	case *messaging.TransactionMessage:
+		if !m.GetActiveGlobals().ExecutorVersion.V2BaikonurEnabled() {
+			break
+		}
+
+		// SyntheticForwardTransaction is the only synthetic transaction type
+		// that does not satisfy this interface, but they are not used in v2
+		txn, ok := msg.Transaction.Body.(protocol.SyntheticTransaction)
+		if !ok {
+			slog.ErrorCtx(m.Context, "Synthetic transaction is not synthetic", "id", msg.ID())
+			break
+		}
+
+		txn.SetIndex(m.syntheticCount)
+		p.Message = msg.Copy() // Clear memoized hashes
+	}
+
+	if dest == nil {
+		panic("nil destination for produced message")
+	}
+	m.produced = append(m.produced, p)
 
 	err := batch.Message(m.message.Hash()).Produced().Add(msg.ID())
 	if err != nil {

internal/core/execute/v2/block/msg_transaction.go

@@ -394,9 +394,9 @@ func (x TransactionMessage) executeTransaction(batch *database.Batch, ctx *Trans
 
 func (x TransactionMessage) postProcess(batch *database.Batch, ctx *TransactionContext, state *chain.ProcessTransactionState, delivered bool) error {
 	// Calculate refunds
-	var swos []protocol.SynthTxnWithOrigin
+	var swos []protocol.SyntheticTransaction
 	for _, newTxn := range state.ProducedTxns {
-		if swo, ok := newTxn.Body.(protocol.SynthTxnWithOrigin); ok {
+		if swo, ok := newTxn.Body.(protocol.SyntheticTransaction); ok {
 			swos = append(swos, swo)
 		}
 	}

internal/core/execute/v2/block/synthetic.go

@@ -60,7 +60,7 @@ func (x *Executor) produceSynthetic(batch *database.Batch, produced []*ProducedM
 // transaction. setSyntheticOrigin sets the refund amount for each synthetic
 // transaction, spreading the potential refund across all produced synthetic
 // transactions.
-func (x *Executor) setSyntheticOrigin(batch *database.Batch, from *protocol.Transaction, produced []protocol.SynthTxnWithOrigin) error {
+func (x *Executor) setSyntheticOrigin(batch *database.Batch, from *protocol.Transaction, produced []protocol.SyntheticTransaction) error {
 	for _, swo := range produced {
 		swo.SetCause(from.ID().Hash(), from.ID().Account())
 	}

internal/core/execute/v2/block/transaction.go

@@ -584,7 +584,7 @@ func (x *TransactionContext) recordFailedTransaction(batch *database.Batch, deli
 	}
 
 	// If this transaction is a synthetic transaction, send a refund
-	if swo, ok := delivery.Transaction.Body.(protocol.SynthTxnWithOrigin); ok {
+	if swo, ok := delivery.Transaction.Body.(protocol.SyntheticTransaction); ok {
 		init, refundAmount := swo.GetRefund()
 		if refundAmount > 0 {
 			refund := new(protocol.SyntheticDepositCredits)

protocol/synthetic_origin.go

@@ -1,4 +1,4 @@
-// Copyright 2022 The Accumulate Authors
+// Copyright 2023 The Accumulate Authors
 //
 // Use of this source code is governed by an MIT-style
 // license that can be found in the LICENSE file or at
@@ -8,15 +8,25 @@ package protocol
 
 import "gitlab.com/accumulatenetwork/accumulate/pkg/url"
 
+type SyntheticTransaction interface {
+	SynthTxnWithOrigin
+	TransactionBody
+}
+
 type SynthTxnWithOrigin interface {
 	GetCause() (cause [32]byte, source *url.URL)
 	SetCause(cause [32]byte, source *url.URL)
 	GetRefund() (initiator *url.URL, refund Fee)
 	SetRefund(initiator *url.URL, refund Fee)
+	SetIndex(index uint64)
 }
 
 var _ SynthTxnWithOrigin = (*SyntheticOrigin)(nil)
 var _ SynthTxnWithOrigin = (*SyntheticCreateIdentity)(nil)
+var _ SynthTxnWithOrigin = (*SyntheticWriteData)(nil)
+var _ SynthTxnWithOrigin = (*SyntheticDepositTokens)(nil)
+var _ SynthTxnWithOrigin = (*SyntheticDepositCredits)(nil)
+var _ SynthTxnWithOrigin = (*SyntheticBurnTokens)(nil)
 
 func (so *SyntheticOrigin) Source() *url.URL {
 	if so.Cause == nil {
@@ -41,3 +51,7 @@ func (so *SyntheticOrigin) SetRefund(initiator *url.URL, refund Fee) {
 	so.Initiator = initiator
 	so.FeeRefund = uint64(refund)
 }
+
+func (so *SyntheticOrigin) SetIndex(index uint64) {
+	so.Index = index
+}

protocol/synthetic_transactions.yml

@@ -16,6 +16,8 @@ SyntheticOrigin:
     - name: FeeRefund
       description: is portion of the cause's fee that will be refunded if this transaction fails
       type: uint
+    - name: Index
+      type: uint
 
 SyntheticCreateIdentity:
   union: { type: transaction }
@@ -52,7 +54,7 @@ SyntheticDepositTokens:
     - name: IsRefund
       type: bool
 
-      
+
 SyntheticDepositCredits:
   union: { type: transaction }
   embeddings:

protocol/transaction.go

@@ -182,10 +182,6 @@ func (t *Transaction) GetAdditionalAuthorities() []*url.URL {
 	return nil
 }
 
-type SyntheticTransaction interface {
-	TransactionBody
-}
-
 func (tx *SendTokens) AddRecipient(to *url.URL, amount *big.Int) {
 	recipient := new(TokenRecipient)
 	recipient.Url = to