refactor(recovery): return RecoveryClaim from ClaimPendingTransactions (#1715)

Signed-off-by: Evan <evanyan@sign.global>

Author: EvanYan1024

token/services/auditor/mock/audit_transaction_store.go

@@ -618,6 +618,6 @@ type transactionDB interface {
 	GetTokenRequest(ctx context.Context, txID string) ([]byte, error)
 	SetStatus(ctx context.Context, txID string, status storage.TxStatus, message string) error
 	AcquireRecoveryLeadership(ctx context.Context, lockID int64) (recovery.Leadership, bool, error)
-	ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*ttxdb.TransactionRecord, error)
+	ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*ttxdb.RecoveryClaim, error)
 	ReleaseRecoveryClaim(ctx context.Context, txID string, owner string, message string) error
 }

token/services/network/fabric/network.go

@@ -103,6 +103,10 @@ type MovementRecord = dbdriver.MovementRecord
 // in that action.
 type TransactionRecord = dbdriver.TransactionRecord
 
+// RecoveryClaim is the minimal projection of a pending transaction row
+// returned by ClaimPendingTransactions for recovery processing.
+type RecoveryClaim = dbdriver.RecoveryClaim
+
 // QueryTransactionsParams defines the parameters for querying movements
 type QueryTransactionsParams = dbdriver.QueryTransactionsParams
 
@@ -353,7 +357,9 @@ func (d *StoreService) AcquireRecoveryLeadership(ctx context.Context, lockID int
 }
 
 // ClaimPendingTransactions returns a claimed batch of Pending transactions older than the given duration.
-func (d *StoreService) ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*TransactionRecord, error) {
+// Each returned RecoveryClaim carries the TxID and StoredAt timestamp the recovery loop needs;
+// the rest of the row is intentionally not projected from SQL.
+func (d *StoreService) ClaimPendingTransactions(ctx context.Context, olderThan time.Duration, leaseDuration time.Duration, limit int, owner string) ([]*RecoveryClaim, error) {
 	storedBefore := time.Now().UTC().Add(-olderThan)
 	logger.DebugfContext(ctx, "claiming pending transactions stored before %s (older than %s), lease duration [%s], limit [%d], owner [%s]",
 		storedBefore, olderThan, leaseDuration, limit, owner)

token/services/storage/auditdb/store.go

@@ -55,7 +55,9 @@ type AuditTransactionStore interface {
 
 	// ClaimPendingTransactions atomically claims a batch of Pending transactions for recovery processing.
 	// Transactions whose recovery lease expired are eligible again.
-	ClaimPendingTransactions(ctx context.Context, params RecoveryClaimParams) ([]*TransactionRecord, error)
+	// Returns the minimal projection (TxID + StoredAt) needed by the recovery loop;
+	// callers do not need the full TransactionRecord.
+	ClaimPendingTransactions(ctx context.Context, params RecoveryClaimParams) ([]*RecoveryClaim, error)
 
 	// ReleaseRecoveryClaim clears the recovery claim metadata for the given transaction if owned by owner.
 	// The message parameter is stored for audit/debugging purposes.

token/services/storage/db/driver/audit.go

@@ -95,7 +95,9 @@ type TransactionStore interface {
 
 	// ClaimPendingTransactions atomically claims a batch of Pending transactions for recovery processing.
 	// Transactions whose recovery lease expired are eligible again.
-	ClaimPendingTransactions(ctx context.Context, params RecoveryClaimParams) ([]*TransactionRecord, error)
+	// Returns the minimal projection (TxID + StoredAt) needed by the recovery loop;
+	// callers do not need the full TransactionRecord.
+	ClaimPendingTransactions(ctx context.Context, params RecoveryClaimParams) ([]*RecoveryClaim, error)
 
 	// ReleaseRecoveryClaim clears the recovery claim metadata for the given transaction if owned by owner.
 	// The message parameter is stored for audit/debugging purposes.
@@ -124,6 +126,20 @@ type RecoveryClaimParams struct {
 	Owner         string
 }
 
+// RecoveryClaim is the minimal projection of a pending transaction row
+// returned by ClaimPendingTransactions. The recovery loop only needs the
+// TxID to act on and the StoredAt timestamp to decide grace-period
+// promotions; the rest of TransactionRecord (action type, amounts,
+// metadata, ...) was always discarded by the caller, so the SQL layer
+// stops projecting it.
+type RecoveryClaim struct {
+	// TxID is the transaction ID claimed for recovery.
+	TxID string
+	// StoredAt is the storage timestamp of the underlying row (UTC), used
+	// by the recovery loop to compute row age for grace-period decisions.
+	StoredAt time.Time
+}
+
 // TransactionRecordReference contains the primary key fields of a transaction request record.
 type TransactionRecordReference struct {
 	// TxID is the unique identifier of the transaction request.

token/services/storage/db/driver/ttx.go

@@ -327,23 +327,19 @@ func (db *TransactionStore) AcquireRecoveryLeadership(ctx context.Context, lockI
 
 // ClaimPendingTransactions returns a claimed batch of Pending transactions.
 // The default SQL implementation is permissive and does not persist recovery claims.
-func (db *TransactionStore) ClaimPendingTransactions(ctx context.Context, params dbdriver.RecoveryClaimParams) ([]*dbdriver.TransactionRecord, error) {
-	transactionsTable, requestsTable := q.Table(db.table.Transactions), q.Table(db.table.Requests)
+// tx_id and stored_at are projected directly from the requests table — the
+// transactions table is no longer joined since it carries no information
+// the recovery loop needs and adding it would re-introduce a fan-out by
+// movement/output that the caller would have to dedupe by tx_id.
+func (db *TransactionStore) ClaimPendingTransactions(ctx context.Context, params dbdriver.RecoveryClaimParams) ([]*dbdriver.RecoveryClaim, error) {
 	query, args := q.Select().
-		Fields(
-			transactionsTable.Field("tx_id"), common3.FieldName("action_type"), common3.FieldName("sender_eid"),
-			common3.FieldName("recipient_eid"), common3.FieldName("token_type"), common3.FieldName("amount"),
-			requestsTable.Field("status"), requestsTable.Field("application_metadata"),
-			requestsTable.Field("public_metadata"), transactionsTable.Field("stored_at"),
-		).
-		From(transactionsTable.Join(requestsTable,
-			cond.Cmp(transactionsTable.Field("tx_id"), "=", requestsTable.Field("tx_id"))),
-		).
+		FieldsByName("tx_id", "stored_at").
+		From(q.Table(db.table.Requests)).
 		Where(cond.And(
 			cond.Eq("status", dbdriver.Pending),
-			cond.Lt(common3.FieldName(db.table.Transactions+".stored_at"), params.OlderThan),
+			cond.Lt("stored_at", params.OlderThan),
 		)).
-		OrderBy(q.Asc(transactionsTable.Field("stored_at"))).
+		OrderBy(q.Asc(common3.FieldName("stored_at"))).
 		Limit(params.Limit).
 		Format(db.ci)
 
@@ -353,19 +349,8 @@ func (db *TransactionStore) ClaimPendingTransactions(ctx context.Context, params
 		return nil, err
 	}
 
-	results := common.NewIterator(rows, func(r *dbdriver.TransactionRecord) error {
-		var amount BigInt
-		var appMeta []byte
-		var pubMeta []byte
-		if err := rows.Scan(&r.TxID, &r.ActionType, &r.SenderEID, &r.RecipientEID, &r.TokenType, &amount, &r.Status, &appMeta, &pubMeta, &r.Timestamp); err != nil {
-			return err
-		}
-		r.Amount = amount.Int
-
-		return errors2.Join(
-			unmarshal(appMeta, &r.ApplicationMetadata),
-			unmarshal(pubMeta, &r.PublicMetadata),
-		)
+	results := common.NewIterator(rows, func(r *dbdriver.RecoveryClaim) error {
+		return rows.Scan(&r.TxID, &r.StoredAt)
 	})
 
 	return iterators.ReadAllPointers(results)

token/services/storage/db/sql/common/transactions.go

@@ -8,6 +8,7 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 	"math/big"
 	"testing"
 	"time"
@@ -49,8 +50,10 @@ func TestClaimPendingTransactions_Atomic(t *testing.T) {
 	oldTime := now.Add(-10 * time.Minute)
 
 	// Add 5 pending transactions
+	txIDs := make([]string, 0, 5)
 	for i := range 5 {
 		txID := "tx" + string(rune('1'+i))
+		txIDs = append(txIDs, txID)
 		err = aw.AddTokenRequest(ctx, txID, []byte("request"), nil, nil, []byte("hash"))
 		require.NoError(t, err)
 
@@ -68,6 +71,7 @@ func TestClaimPendingTransactions_Atomic(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store1, oldTime, txIDs...)
 
 	// Both instances try to claim the same transactions
 	params := tokensdriver.RecoveryClaimParams{
@@ -128,6 +132,7 @@ func TestClaimPendingTransactions_Lease(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store, oldTime, txID)
 
 	// Claim with very short lease
 	params := tokensdriver.RecoveryClaimParams{
@@ -196,6 +201,7 @@ func TestClaimPendingTransactions_Idempotent(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store, oldTime, txID)
 
 	// Claim transaction
 	params := tokensdriver.RecoveryClaimParams{
@@ -238,8 +244,10 @@ func TestClaimPendingTransactions_Limit(t *testing.T) {
 	now := time.Now().UTC()
 	oldTime := now.Add(-10 * time.Minute)
 
+	txIDs := make([]string, 0, 10)
 	for i := range 10 {
 		txID := "tx" + string(rune('0'+i))
+		txIDs = append(txIDs, txID)
 		err = aw.AddTokenRequest(ctx, txID, []byte("request"), nil, nil, []byte("hash"))
 		require.NoError(t, err)
 
@@ -257,6 +265,9 @@ func TestClaimPendingTransactions_Limit(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	for i, txID := range txIDs {
+		ageRequests(t, ctx, store, oldTime.Add(time.Duration(i)*time.Second), txID)
+	}
 
 	// Claim with limit of 3
 	params := tokensdriver.RecoveryClaimParams{
@@ -317,6 +328,7 @@ func TestReleaseRecoveryClaim(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store, oldTime, txID)
 
 	// Claim transaction
 	params := tokensdriver.RecoveryClaimParams{
@@ -381,6 +393,7 @@ func TestReleaseRecoveryClaim_WrongOwner(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store, oldTime, txID)
 
 	// Claim transaction
 	params := tokensdriver.RecoveryClaimParams{
@@ -427,8 +440,10 @@ func TestCleanupExpiredClaims(t *testing.T) {
 	now := time.Now().UTC()
 	oldTime := now.Add(-10 * time.Minute)
 
+	txIDs := make([]string, 0, 3)
 	for i := range 3 {
 		txID := "tx" + string(rune('1'+i))
+		txIDs = append(txIDs, txID)
 		err = aw.AddTokenRequest(ctx, txID, []byte("request"), nil, nil, []byte("hash"))
 		require.NoError(t, err)
 
@@ -446,6 +461,7 @@ func TestCleanupExpiredClaims(t *testing.T) {
 
 	err = aw.Commit()
 	require.NoError(t, err)
+	ageRequests(t, ctx, store, oldTime, txIDs...)
 
 	// Claim with very short lease
 	params := tokensdriver.RecoveryClaimParams{
@@ -473,3 +489,18 @@ func TestCleanupExpiredClaims(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, claimed, 3, "Should be able to claim after cleanup")
 }
+
+func ageRequests(t *testing.T, ctx context.Context, store *TransactionStore, storedAt time.Time, txIDs ...string) {
+	t.Helper()
+
+	// #nosec G201 -- table name comes from the test-created store.
+	query := fmt.Sprintf("UPDATE %s SET stored_at = $1 WHERE tx_id = $2", store.tables.Requests)
+	for _, txID := range txIDs {
+		result, err := store.writeDB.ExecContext(ctx, query, storedAt, txID)
+		require.NoError(t, err)
+
+		rowsAffected, err := result.RowsAffected()
+		require.NoError(t, err)
+		require.EqualValues(t, 1, rowsAffected)
+	}
+}