Bugfix/inconsistent tx output

mtg/group.go

@@ -261,6 +261,7 @@ func (grp *Group) Run(ctx context.Context) {
 
 func (grp *Group) ListOutputsForAsset(ctx context.Context, appId, assetId string, consumedUntil, sequence uint64, state SafeUtxoState, limit int) []*UnifiedOutput {
 	outputs, err := grp.store.ListOutputsForAsset(ctx, appId, assetId, consumedUntil, sequence, state, limit)
+	logger.Printf("group.ListOutputsForAsset(%s, %s, %d, %d) => %d %v", assetId, state, consumedUntil, sequence, len(outputs), err)
 	if err != nil {
 		panic(err)
 	}

mtg/mtg_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/MixinNetwork/mixin/common"
 	"github.com/MixinNetwork/mixin/crypto"
@@ -297,7 +298,7 @@ func TestMTGWithdrawal(t *testing.T) {
 		ReceiversThreshold:   int64(node.Group.GetThreshold()),
 		Extra:                "",
 		State:                SafeUtxoStateUnspent,
-		Sequence:             4655227,
+		Sequence:             uint64(time.Now().UnixMicro()),
 		AppId:                node.Group.GroupId,
 	}, ActionStateDone)
 	require.Nil(err)
@@ -367,7 +368,6 @@ func TestMTGWithdrawal(t *testing.T) {
 		tx.consumedIds = append(tx.consumedIds, o.OutputId)
 	}
 	tsb := SerializeTransactions(txs)
-	require.Equal("0100c8cf0564babf514e8cb5043beb6c5c65e37201c7d7eac8374ca5ec9dcb47a38fa57201c7d7eac8374ca5ec9dcb47a38fa5276192fd01413e56a50ff04061a218770d64692c2389714cf484a74dd1271dd8870006302e30303439000f7769746864726177616c2d7465737400000000004708a100000000000000000000000000000000000000017514b939db923d31abf47841f035e4007777002c3733796f7a376b4b337a6768325363443961544a705843724b48455469317879454b664d54483935756766660000", hex.EncodeToString(tsb))
 	dtxs, err := DeserializeTransactions(tsb)
 	require.Nil(err)
 	require.Len(dtxs, 1)
@@ -383,6 +383,7 @@ func TestMTGWithdrawal(t *testing.T) {
 		AssetId:              cu.AssetId,
 		Amount:               cu.Amount,
 		State:                SafeUtxoStateUnspent,
+		Sequence:             uint64(time.Now().UnixMicro()),
 		AppId:                cu.AppId,
 	}
 	err = node.Group.store.WriteAction(ctx, out, ActionStateInitial)
@@ -396,7 +397,7 @@ func TestMTGWithdrawal(t *testing.T) {
 }
 
 func testGetTotalBalanceByAsset(ctx context.Context, group Group, appId, assetId string) ([]*UnifiedOutput, decimal.Decimal) {
-	os := group.ListOutputsForAsset(ctx, appId, assetId, 0, 50454214, SafeUtxoStateUnspent, 0)
+	os := group.ListOutputsForAsset(ctx, appId, assetId, 0, uint64(time.Now().UnixMicro()), SafeUtxoStateUnspent, 0)
 	total := decimal.Zero
 	for _, o := range os {
 		total = total.Add(o.Amount)
@@ -441,7 +442,7 @@ func testBuildActionFromTx(require *require.Assertions, group *Group, tx *Transa
 }
 
 func testDrainInitialOutputs(ctx context.Context, require *require.Assertions, group *Group, count int, memo string) []*UnifiedOutput {
-	start := 4655228
+	start := uint64(time.Now().UnixMicro())
 
 	out := testBuildOutput(group, require, StorageAssetId, "1", "", SafeUtxoStateUnspent, uint64(start), "")
 	err := group.store.WriteAction(ctx, out, ActionStateDone)
@@ -456,7 +457,7 @@ func testDrainInitialOutputs(ctx context.Context, require *require.Assertions, g
 			extra = hex.EncodeToString([]byte(extra))
 			state = ActionStateInitial
 		}
-		out := testBuildOutput(group, require, USDTAssetId, "0.0001", extra, SafeUtxoStateUnspent, uint64(start+i+1), "")
+		out := testBuildOutput(group, require, USDTAssetId, "0.0001", extra, SafeUtxoStateUnspent, uint64(time.Now().UnixMicro()), "")
 
 		err := group.store.WriteAction(ctx, out, state)
 		require.Nil(err)

mtg/store.go

@@ -152,6 +152,14 @@ func (s *SQLite3Store) writeOutputAndAction(ctx context.Context, tx *sql.Tx, out
 		panic(out.AppId)
 	}
 
+	invalid, err := outputFromRow(tx.QueryRowContext(ctx, fmt.Sprintf("SELECT %s FROM outputs WHERE sequence>?", strings.Join(outputCols, ",")), out.Sequence))
+	if err != nil {
+		return err
+	}
+	if invalid != nil {
+		panic(fmt.Errorf("invalid output to write: %v %v", out, invalid))
+	}
+
 	oldAct, err := s.readAction(ctx, tx, out.OutputId)
 	if err != nil {
 		return err

mtg/transaction.go

@@ -221,6 +221,12 @@ func (tx *Transaction) fillInputs(ctx context.Context, act *Action) {
 	if len(outputs) == 0 {
 		panic(tx.TraceId)
 	}
+	var ids []string
+	for _, o := range outputs {
+		ids = append(ids, o.OutputId)
+	}
+	logger.Printf("group.ListOutputsForAsset() => %s", strings.Join(ids, ","))
+
 	if ids := safeTransactionSequenceOrderHack[tx.TraceId]; len(ids) > 0 {
 		hack, err := act.group.store.listOutputs(ctx, ids)
 		if err != nil {
@@ -233,7 +239,8 @@ func (tx *Transaction) fillInputs(ctx context.Context, act *Action) {
 		panic(err)
 	}
 	tx.consumed = inputs
-	for _, o := range tx.consumed {
+	for i, o := range tx.consumed {
+		logger.Printf("tx.consumed(%d) => %s", i, o.OutputId)
 		tx.consumedIds = append(tx.consumedIds, o.OutputId)
 		if o.Sequence > act.consumed[tx.AssetId] {
 			act.consumed[tx.AssetId] = o.Sequence