deposit arrives ealier than observer confirmations

solana/mvm.go

@@ -743,7 +743,7 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 		logger.Printf("solana.ExtractTransferFromTransactionByIndex(%s %s %d) => %v", out.OutputId, out.DepositHash.String, out.DepositIndex.Int64, t)
 	}
 	if t == nil || t.AssetId != out.AssetId || t.Receiver != node.SolanaDepositEntry().String() {
-		return node.failDepositRequest(ctx, out, "")
+		return node.failDepositRequest(ctx, out, "", false)
 	}
 	asset, err := common.SafeReadAssetUntilSufficient(ctx, t.AssetId)
 	if err != nil {
@@ -758,7 +758,7 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 	actual := mc.NewIntegerFromString(out.Amount.String())
 	if expected.Cmp(actual) != 0 {
 		logger.Printf("invalid deposit amount: %s %s", actual.String(), out.Amount.String())
-		return node.failDepositRequest(ctx, out, "")
+		return node.failDepositRequest(ctx, out, "", false)
 	}
 
 	// user == nil: transfer solana withdrawn assets from mtg to mtg deposit entry by post call for failed prepare call
@@ -773,15 +773,15 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 		memo := solanaApp.ExtractMemoFromTransaction(ctx, tx, meta, node.SolanaPayer())
 		logger.Printf("solana.ExtractMemoFromTransaction(%s) => %s", tx.Signatures[0].String(), memo)
 		if memo == "" {
-			return node.failDepositRequest(ctx, out, "")
+			return node.failDepositRequest(ctx, out, "", false)
 		}
 		call, err = node.store.ReadSystemCallByRequestId(ctx, memo, common.RequestStateFailed)
 		logger.Printf("store.ReadSystemCallByRequestId(%s) => %v %v", memo, call, err)
 		if err != nil {
 			panic(err)
 		}
 		if call == nil || call.Type != store.CallTypePrepare {
-			return node.failDepositRequest(ctx, out, "")
+			return node.failDepositRequest(ctx, out, "", false)
 		}
 		superior, err := node.store.ReadSystemCallByRequestId(ctx, call.Superior, common.RequestStateFailed)
 		logger.Printf("store.ReadSystemCallByRequestId(%s) => %v %v", call.Superior, superior, err)
@@ -800,7 +800,7 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 			panic(err)
 		}
 		if call == nil || call.State != common.RequestStateDone {
-			return node.failDepositRequest(ctx, out, "")
+			return node.failDepositRequest(ctx, out, "", true)
 		}
 		switch call.Type {
 		case store.CallTypeDeposit:
@@ -811,7 +811,7 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 			}
 			call = superior
 		default:
-			return node.failDepositRequest(ctx, out, "")
+			return node.failDepositRequest(ctx, out, "", false)
 		}
 	}
 	mix, err := bot.NewMixAddressFromString(user.MixAddress)
@@ -822,7 +822,7 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 	id = common.UniqueId(id, t.Receiver)
 	mtx := node.buildTransaction(ctx, out, node.conf.AppId, t.AssetId, mix.Members(), int(mix.Threshold), out.Amount.String(), []byte(out.DepositHash.String), id)
 	if mtx == nil {
-		return node.failDepositRequest(ctx, out, t.AssetId)
+		return node.failDepositRequest(ctx, out, t.AssetId, false)
 	}
 	txs := []*mtg.Transaction{mtx}
 	old := call.GetRefundIds()
@@ -838,9 +838,9 @@ func (node *Node) processDeposit(ctx context.Context, out *mtg.Action, restored
 	return txs, ""
 }
 
-func (node *Node) failDepositRequest(ctx context.Context, out *mtg.Action, compaction string) ([]*mtg.Transaction, string) {
+func (node *Node) failDepositRequest(ctx context.Context, out *mtg.Action, compaction string, save bool) ([]*mtg.Transaction, string) {
 	logger.Printf("node.failDepositRequest(%v %s)", out, compaction)
-	err := node.store.FailDepositRequestIfNotExist(ctx, out, compaction)
+	err := node.store.FailDepositRequestIfNotExist(ctx, out, compaction, save)
 	if err != nil {
 		panic(err)
 	}
@@ -1070,11 +1070,28 @@ func (node *Node) confirmBurnRelatedSystemCall(ctx context.Context, req *store.R
 		txs = append(txs, tx)
 		ids = append(ids, tx.TraceId)
 	}
+
+	fd, err := node.store.ReadFailedDepositByHash(ctx, signature)
+	if err != nil {
+		panic(err)
+	}
+	if fd != nil {
+		id := common.UniqueId(fd.Hash, fmt.Sprint(fd.Index))
+		id = common.UniqueId(id, node.SolanaDepositEntry().String())
+		memo := []byte(fd.Hash)
+		tx := node.buildTransaction(ctx, req.Output, node.conf.AppId, fd.AssetId, mix.Members(), int(mix.Threshold), fd.Amount, memo, id)
+		if tx == nil {
+			return node.failRequest(ctx, req, fd.AssetId)
+		}
+		txs = append(txs, tx)
+		ids = append(ids, tx.TraceId)
+	}
+
 	old := call.GetRefundIds()
 	old = append(old, ids...)
 	call.RefundTraces = sql.NullString{Valid: true, String: strings.Join(old, ",")}
 
-	err = node.store.ConfirmBurnRelatedSystemCallWithRequest(ctx, req, call, txs)
+	err = node.store.ConfirmBurnRelatedSystemCallWithRequest(ctx, req, call, fd, txs)
 	if err != nil {
 		panic(err)
 	}

store/call.go

@@ -276,7 +276,7 @@ func (s *SQLite3Store) ConfirmSystemCallsWithRequest(ctx context.Context, req *R
 	return tx.Commit()
 }
 
-func (s *SQLite3Store) ConfirmBurnRelatedSystemCallWithRequest(ctx context.Context, req *Request, call *SystemCall, txs []*mtg.Transaction) error {
+func (s *SQLite3Store) ConfirmBurnRelatedSystemCallWithRequest(ctx context.Context, req *Request, call *SystemCall, deposit *UnconfirmedDeposit, txs []*mtg.Transaction) error {
 	switch call.Type {
 	case CallTypePostProcess, CallTypeDeposit:
 	default:
@@ -304,6 +304,13 @@ func (s *SQLite3Store) ConfirmBurnRelatedSystemCallWithRequest(ctx context.Conte
 		return fmt.Errorf("SQLite3Store UPDATE system_calls %v", err)
 	}
 
+	if deposit != nil {
+		err = s.handleFailedDepositByRequest(ctx, tx, deposit, req)
+		if err != nil {
+			return err
+		}
+	}
+
 	err = s.finishRequest(ctx, tx, req, txs, "")
 	if err != nil {
 		return err

store/request.go

@@ -146,7 +146,7 @@ func (s *SQLite3Store) WriteDepositRequestIfNotExist(ctx context.Context, out *m
 	return tx.Commit()
 }
 
-func (s *SQLite3Store) FailDepositRequestIfNotExist(ctx context.Context, out *mtg.Action, compaction string) error {
+func (s *SQLite3Store) FailDepositRequestIfNotExist(ctx context.Context, out *mtg.Action, compaction string, save bool) error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -167,6 +167,13 @@ func (s *SQLite3Store) FailDepositRequestIfNotExist(ctx context.Context, out *mt
 		return fmt.Errorf("INSERT requests %v", err)
 	}
 
+	if save {
+		err = s.writeFailedDeposit(ctx, tx, out)
+		if err != nil {
+			return err
+		}
+	}
+
 	err = s.writeActionResult(ctx, tx, out.OutputId, compaction, nil, out.OutputId)
 	if err != nil {
 		return err

store/schema.sql

@@ -239,6 +239,20 @@ CREATE TABLE IF NOT EXISTS failed_calls (
 );
 
 
+CREATE TABLE IF NOT EXISTS unconfirmed_deposits (
+  output_id     VARCHAR NOT NULL,
+  mixin_hash    VARCHAR NOT NULL,
+  mixin_index   INTEGER NOT NULL,
+  asset_id      VARCHAR NOT NULL,
+  amount        VARCHAR NOT NULL,
+  handled_by    VARCHAR,
+  created_at    TIMESTAMP NOT NULL,
+  PRIMARY KEY ('output_id')
+);
+
+CREATE INDEX IF NOT EXISTS unconfirmed_deposits_by_hash ON unconfirmed_deposits(mixin_hash);
+
+
 CREATE TABLE IF NOT EXISTS burn_system_calls (
   call_id       VARCHAR NOT NULL,
   asset_id      VARCHAR NOT NULL,

store/unconfirmed_deposit.go

@@ -0,0 +1,61 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/MixinNetwork/safe/mtg"
+)
+
+type UnconfirmedDeposit struct {
+	OutputId  string
+	Hash      string
+	Index     int64
+	AssetId   string
+	Amount    string
+	HandledBy sql.NullString
+	CreatedAt time.Time
+}
+
+var unconfirmedDepositCols = []string{"output_id", "mixin_hash", "mixin_index", "asset_id", "amount", "handled_by", "created_at"}
+
+func unconfirmedDepositFromRow(row Row) (*UnconfirmedDeposit, error) {
+	var d UnconfirmedDeposit
+	err := row.Scan(&d.OutputId, &d.Hash, &d.Index, &d.AssetId, &d.Amount, &d.HandledBy, &d.CreatedAt)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	return &d, err
+}
+
+func (s *SQLite3Store) writeFailedDeposit(ctx context.Context, tx *sql.Tx, out *mtg.Action) error {
+	existed, err := s.checkExistence(ctx, tx, "SELECT output_id FROM unconfirmed_deposits WHERE output_id=?", out.OutputId)
+	if err != nil || existed {
+		return err
+	}
+
+	vals := []any{out.OutputId, out.DepositHash.String, out.DepositIndex, out.AssetId, out.Amount.String(), nil, out.SequencerCreatedAt}
+	err = s.execOne(ctx, tx, buildInsertionSQL("unconfirmed_deposits", unconfirmedDepositCols), vals...)
+	if err != nil {
+		return fmt.Errorf("INSERT unconfirmed_deposits %v", err)
+	}
+	return nil
+}
+
+func (s *SQLite3Store) handleFailedDepositByRequest(ctx context.Context, tx *sql.Tx, d *UnconfirmedDeposit, req *Request) error {
+	err := s.execOne(ctx, tx, "UPDATE unconfirmed_deposits SET handled_by=? WHERE output_id=? AND handled_by IS NULL", req.Id, d.OutputId)
+	if err != nil {
+		return fmt.Errorf("UPDATE unconfirmed_deposits %v", err)
+	}
+	return nil
+}
+
+func (s *SQLite3Store) ReadFailedDepositByHash(ctx context.Context, hash string) (*UnconfirmedDeposit, error) {
+	query := fmt.Sprintf("SELECT %s FROM unconfirmed_deposits WHERE mixin_hash=?", strings.Join(unconfirmedDepositCols, ","))
+	row := s.db.QueryRowContext(ctx, query, hash)
+
+	return unconfirmedDepositFromRow(row)
+}