diff --git a/token/services/storage/db/sql/common/tokenlock_test_utils.go b/token/services/storage/db/sql/common/tokenlock_test_utils.go index 5997e25213..8c25150a10 100644 --- a/token/services/storage/db/sql/common/tokenlock_test_utils.go +++ b/token/services/storage/db/sql/common/tokenlock_test_utils.go @@ -7,10 +7,13 @@ SPDX-License-Identifier: Apache-2.0 package common import ( + "context" "database/sql" "testing" + "time" "github.com/DATA-DOG/go-sqlmock" + fscerrors "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" "github.com/hyperledger-labs/fabric-token-sdk/token/token" "github.com/onsi/gomega" ) @@ -54,3 +57,54 @@ func TestUnlockByTxID(t *testing.T, store tokenLockStoreConstructor) { gomega.Expect(mockDB.ExpectationsWereMet()).To(gomega.Succeed()) gomega.Expect(err).ToNot(gomega.HaveOccurred()) } + +// TestLockContextCancelled verifies that Lock propagates context cancellation +// from ExecContext back to the caller as a context error. +func TestLockContextCancelled(t *testing.T, store tokenLockStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + tokenID := token.ID{TxId: "1234", Index: 5} + trID := "5555" + now := sqlmock.AnyArg() + + // The mock will block for 1 s; the context expires after 10 ms. + mockDB. + ExpectExec("INSERT INTO TOKEN_LOCKS \\(consumer_tx_id, tx_id, idx, created_at\\) VALUES \\(\\$1, \\$2, \\$3, \\$4\\)"). + WithArgs(trID, tokenID.TxId, tokenID.Index, now). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err = store(db).Lock(ctx, &tokenID, trID) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestUnlockByTxIDContextCancelled verifies that UnlockByTxID propagates context +// cancellation from ExecContext back to the caller as a context error. +func TestUnlockByTxIDContextCancelled(t *testing.T, store tokenLockStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + consumerTxID := "1234" + + mockDB. + ExpectExec("DELETE FROM TOKEN_LOCKS WHERE consumer_tx_id = \\$1"). + WithArgs(consumerTxID). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err = store(db).UnlockByTxID(ctx, consumerTxID) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} diff --git a/token/services/storage/db/sql/common/tokens.go b/token/services/storage/db/sql/common/tokens.go index d3ea7360d3..e880aa7098 100644 --- a/token/services/storage/db/sql/common/tokens.go +++ b/token/services/storage/db/sql/common/tokens.go @@ -116,6 +116,7 @@ func (db *TokenStore) DeleteTokens(ctx context.Context, deletedBy string, ids .. Where(HasTokens("tx_id", "idx", ids...)). Format(db.ci) logging.Debug(logger, query, args) + if _, err := db.writeDB.ExecContext(ctx, query, args...); err != nil { return errors.Wrapf(err, "error setting tokens to deleted [%v]", ids) } @@ -1301,6 +1302,7 @@ func (t *TokenTransaction) Delete(ctx context.Context, tokenID token.ID, deleted Format(t.ci) logging.Debug(logger, query, args) + if _, err := t.tx.ExecContext(ctx, query, args...); err != nil { return errors.Wrapf(err, "error setting token to deleted [%s]", tokenID.TxId) } @@ -1320,6 +1322,7 @@ func (t *TokenTransaction) StoreToken(ctx context.Context, tr driver.TokenRecord OnConflictDoNothing(). Format() logging.Debug(logger, query, args) + if _, err := t.tx.ExecContext(ctx, query, args...); err != nil { logger.Errorf("error storing token [%s] in table [%s] [%s]: [%s][%s]", tr.TxID, t.table.Tokens, query, err, string(debug.Stack())) diff --git a/token/services/storage/db/sql/common/tokens_test_util.go b/token/services/storage/db/sql/common/tokens_test_util.go new file mode 100644 index 0000000000..ea6f0ef9db --- /dev/null +++ b/token/services/storage/db/sql/common/tokens_test_util.go @@ -0,0 +1,45 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package common + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + fscerrors "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + "github.com/hyperledger-labs/fabric-token-sdk/token/token" + "github.com/onsi/gomega" +) + +type tokenStoreConstructor func(*sql.DB) *TokenStore + +// TestDeleteTokensContextCancelled verifies that a cancelled context is propagated +// from ExecContext back to the caller of DeleteTokens as a context error. +func TestDeleteTokensContextCancelled(t *testing.T, store tokenStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + ids := []*token.ID{{TxId: "tx1", Index: 0}} + + // The mock delays the UPDATE by 1 s; the context expires after 10 ms. + mockDB. + ExpectExec("UPDATE"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err = store(db).DeleteTokens(ctx, "spender", ids...) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} diff --git a/token/services/storage/db/sql/common/transactions.go b/token/services/storage/db/sql/common/transactions.go index 88c201d630..6613d54fa4 100644 --- a/token/services/storage/db/sql/common/transactions.go +++ b/token/services/storage/db/sql/common/transactions.go @@ -393,6 +393,7 @@ func (db *TransactionStore) AddTransactionEndorsementAck(ctx context.Context, tx Format() logging.Debug(logger, query, txID, fmt.Sprintf("(%d bytes)", len(endorser)), fmt.Sprintf("(%d bytes)", len(sigma)), now) + if _, err = db.writeDB.ExecContext(ctx, query, args...); err != nil { return ttxDBError(err) } @@ -602,6 +603,7 @@ func (w *TransactionStoreTransaction) AddTransaction(ctx context.Context, rs ... Rows(rows). Format() logging.Debug(logger, query, args) + _, err := w.txn.ExecContext(ctx, query, args...) return ttxDBError(err) @@ -632,6 +634,7 @@ func (w *TransactionStoreTransaction) AddTokenRequest(ctx context.Context, txID Row(txID, tr, dbdriver.Pending, "", ja, jp, ppHash, time.Now().UTC()). Format() logging.Debug(logger, query, txID, fmt.Sprintf("(%d bytes)", len(tr)), len(applicationMetadata), len(publicMetadata), len(ppHash)) + _, err = w.txn.ExecContext(ctx, query, args...) return ttxDBError(err) @@ -666,6 +669,7 @@ func (w *TransactionStoreTransaction) AddMovement(ctx context.Context, rs ...dbd Rows(rows). Format() logging.Debug(logger, query, args) + _, err := w.txn.ExecContext(ctx, query, args...) return ttxDBError(err) diff --git a/token/services/storage/db/sql/common/transactions_test_util.go b/token/services/storage/db/sql/common/transactions_test_util.go index 055aa2d885..3c305382d1 100644 --- a/token/services/storage/db/sql/common/transactions_test_util.go +++ b/token/services/storage/db/sql/common/transactions_test_util.go @@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0 package common import ( + "context" "database/sql" driver2 "database/sql/driver" "math/big" @@ -14,6 +15,7 @@ import ( "time" "github.com/DATA-DOG/go-sqlmock" + fscerrors "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators" token2 "github.com/hyperledger-labs/fabric-token-sdk/token" "github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/db/driver" @@ -425,3 +427,162 @@ func TestAWAddValidationRecord(t *testing.T, store transactionsStoreConstructor) gomega.Expect(mockDB.ExpectationsWereMet()).To(gomega.Succeed()) } + +// TestGetStatusContextCancelled verifies that a cancelled context is propagated from +// QueryContext back to the caller of GetStatus as a context error. +func TestGetStatusContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + // The mock delays the query by 1 s; the context expires after 10 ms. + mockDB. + ExpectQuery("SELECT status, status_message FROM REQUESTS WHERE tx_id = \\$1"). + WithArgs("1234"). + WillDelayFor(time.Second). + WillReturnRows(mockDB.NewRows([]string{"status", "status_message"})) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + _, _, err = store(db).GetStatus(ctx, "1234") + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestAddTransactionEndorsementAckContextCancelled verifies that a cancelled context +// is propagated from ExecContext back to the caller of AddTransactionEndorsementAck. +func TestAddTransactionEndorsementAckContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + mockDB. + ExpectExec("INSERT INTO TRANSACTION_ENDORSE_ACK"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(1, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err = store(db).AddTransactionEndorsementAck(ctx, "txid", token2.Identity([]byte("endorser")), []byte("sigma")) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestSetStatusContextCancelled verifies that a cancelled context is propagated from +// ExecContext back to the caller of SetStatus as a context error. +func TestSetStatusContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + mockDB. + ExpectExec("UPDATE REQUESTS SET status = \\$1, status_message = \\$2 WHERE tx_id = \\$3"). + WithArgs(driver.Confirmed, "message", "txid"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(1, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err = store(db).SetStatus(ctx, "txid", driver.Confirmed, "message") + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestAWAddTransactionContextCancelled verifies that a cancelled context is propagated +// from ExecContext through AddTransaction inside a TransactionStoreTransaction. +func TestAWAddTransactionContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + input := driver.TransactionRecord{ + TxID: "txid", + ActionType: driver.Transfer, + SenderEID: "sender", + RecipientEID: "recipient", + TokenType: "USD", + Amount: big.NewInt(10), + Timestamp: time.Now(), + } + + mockDB.ExpectBegin() + mockDB. + ExpectExec("INSERT INTO TRANSACTIONS"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + aw, err := store(db).NewTransactionStoreTransaction() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + err = aw.AddTransaction(ctx, input) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestAWAddTokenRequestContextCancelled verifies that a cancelled context is propagated +// from ExecContext through AddTokenRequest inside a TransactionStoreTransaction. +func TestAWAddTokenRequestContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + mockDB.ExpectBegin() + mockDB. + ExpectExec("INSERT INTO REQUESTS"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + aw, err := store(db).NewTransactionStoreTransaction() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + err = aw.AddTokenRequest(ctx, "txid", []byte("tr"), nil, nil, []byte("pphash")) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} + +// TestAWAddMovementContextCancelled verifies that a cancelled context is propagated +// from ExecContext through AddMovement inside a TransactionStoreTransaction. +func TestAWAddMovementContextCancelled(t *testing.T, store transactionsStoreConstructor) { + gomega.RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + input := driver.MovementRecord{ + TxID: "txid", + EnrollmentID: "EID", + TokenType: "USD", + Amount: big.NewInt(10), + Status: driver.Pending, + } + + mockDB.ExpectBegin() + mockDB. + ExpectExec("INSERT INTO MOVEMENTS"). + WillDelayFor(time.Second). + WillReturnResult(sqlmock.NewResult(0, 1)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + aw, err := store(db).NewTransactionStoreTransaction() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + err = aw.AddMovement(ctx, input) + + gomega.Expect(fscerrors.Is(err, sqlmock.ErrCancelled)).To(gomega.BeTrue(), + "expected cancellation error, got: %v", err) +} diff --git a/token/services/storage/db/sql/sqlite/tokenlock_test.go b/token/services/storage/db/sql/sqlite/tokenlock_test.go index 204e4e35d7..0578ddd51e 100644 --- a/token/services/storage/db/sql/sqlite/tokenlock_test.go +++ b/token/services/storage/db/sql/sqlite/tokenlock_test.go @@ -57,3 +57,11 @@ func TestLock(t *testing.T) { func TestUnlockByTxID(t *testing.T) { common3.TestUnlockByTxID(t, mockTokenLockStore) } + +func TestLockContextCancelled(t *testing.T) { + common3.TestLockContextCancelled(t, mockTokenLockStore) +} + +func TestUnlockByTxIDContextCancelled(t *testing.T) { + common3.TestUnlockByTxIDContextCancelled(t, mockTokenLockStore) +} diff --git a/token/services/storage/db/sql/sqlite/tokens_test.go b/token/services/storage/db/sql/sqlite/tokens_test.go new file mode 100644 index 0000000000..5440498fe4 --- /dev/null +++ b/token/services/storage/db/sql/sqlite/tokens_test.go @@ -0,0 +1,25 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package sqlite + +import ( + "database/sql" + "testing" + + common2 "github.com/hyperledger-labs/fabric-token-sdk/token/services/storage/db/sql/common" +) + +func mockTokenStore(db *sql.DB) *common2.TokenStore { + tables, _ := common2.GetTableNames("") + store, _ := common2.NewTokenStoreWithNotifier(db, db, tables, NewConditionInterpreter(), nil) + + return store +} + +func TestDeleteTokensContextCancelled(t *testing.T) { + common2.TestDeleteTokensContextCancelled(t, mockTokenStore) +} diff --git a/token/services/storage/db/sql/sqlite/transactions_test.go b/token/services/storage/db/sql/sqlite/transactions_test.go index d32102102d..37e7a0f07d 100644 --- a/token/services/storage/db/sql/sqlite/transactions_test.go +++ b/token/services/storage/db/sql/sqlite/transactions_test.go @@ -81,3 +81,27 @@ func TestAWAddMovement(t *testing.T) { func TestAWAddValidationRecord(t *testing.T) { common2.TestAWAddValidationRecord(t, mockTransactionsStore) } + +func TestGetStatusContextCancelled(t *testing.T) { + common2.TestGetStatusContextCancelled(t, mockTransactionsStore) +} + +func TestAddTransactionEndorsementAckContextCancelled(t *testing.T) { + common2.TestAddTransactionEndorsementAckContextCancelled(t, mockTransactionsStore) +} + +func TestSetStatusContextCancelled(t *testing.T) { + common2.TestSetStatusContextCancelled(t, mockTransactionsStore) +} + +func TestAWAddTransactionContextCancelled(t *testing.T) { + common2.TestAWAddTransactionContextCancelled(t, mockTransactionsStore) +} + +func TestAWAddTokenRequestContextCancelled(t *testing.T) { + common2.TestAWAddTokenRequestContextCancelled(t, mockTransactionsStore) +} + +func TestAWAddMovementContextCancelled(t *testing.T) { + common2.TestAWAddMovementContextCancelled(t, mockTransactionsStore) +} diff --git a/token/services/ttx/collectendorsements_test.go b/token/services/ttx/collectendorsements_test.go index 1cede03679..b4b856c101 100644 --- a/token/services/ttx/collectendorsements_test.go +++ b/token/services/ttx/collectendorsements_test.go @@ -7,12 +7,19 @@ SPDX-License-Identifier: Apache-2.0 package ttx_test import ( + "context" "testing" + "time" "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" + "github.com/hyperledger-labs/fabric-smart-client/platform/view/view" "github.com/hyperledger-labs/fabric-token-sdk/token/services/ttx" "github.com/hyperledger-labs/fabric-token-sdk/token/services/ttx/dep/mock" + jsession "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/json/session" + utilsession "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/session" + sessionmock "github.com/hyperledger-labs/fabric-token-sdk/token/services/utils/session/mock" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // TestCleanupExternalWallets_Success tests that CleanupExternalWallets calls Done() on all wallets @@ -96,3 +103,173 @@ func TestCleanupExternalWallets_EmptyMap(t *testing.T) { // Should not panic with empty map view.CleanupExternalWallets(ctx, externalWallets) } + +// --------------------------------------------------------------------------- +// Timeout / context-cancellation tests for the endorsement-flow receive calls +// --------------------------------------------------------------------------- +// +// CollectEndorsementsView.signRemote and CollectEndorsementsView.distributeTxToParty +// both call TypedSession.ReceiveTypedWithTimeout, which ultimately delegates to +// session.S.ReceiveRawWithTimeout. The select inside that helper simultaneously +// watches the channel for a message, a per-phase timer, and ctx.Done(). +// +// The tests below exercise the contract at that boundary using the same message +// types used by the actual implementation (TypeSignature for signatures / +// acknowledgements, TypeTransaction for the endorsed transaction). + +// TestRemoteSignerTimeoutViaCancelledContext verifies that when the remote signer +// never responds, a cancelled context causes the receive to return immediately +// with ErrContextDone. This mirrors the signRemote call site in +// CollectEndorsementsView. A 1-ms explicit timeout is also tested to confirm +// the timer path. +func TestRemoteSignerTimeoutViaCancelledContext(t *testing.T) { + t.Run("context cancelled", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no message will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "remote-signer-session"}) + + cancelledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + s := utilsession.New(ms, cancelledCtx, jsession.JSONMarshaller{}) + + var sig ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &sig, time.Minute) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrContextDone, + "cancelled context must unblock the receive immediately with ErrContextDone") + }) + + t.Run("per-phase timer fires", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no message will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "remote-signer-session"}) + + s := utilsession.New(ms, context.Background(), jsession.JSONMarshaller{}) + + var sig ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &sig, 1*time.Millisecond) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrTimeout, + "expired per-phase timer must unblock the receive with ErrTimeout") + }) +} + +// TestDistributionAckTimeoutViaCancelledContext verifies that when the receiving +// party never sends the acknowledgement (TypeSignature) after the endorsed +// transaction is distributed, a cancelled context unblocks immediately. +// This mirrors the distributeTxToParty call site in CollectEndorsementsView. +func TestDistributionAckTimeoutViaCancelledContext(t *testing.T) { + t.Run("context cancelled", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no ack will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "distribution-ack-session"}) + + cancelledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + s := utilsession.New(ms, cancelledCtx, jsession.JSONMarshaller{}) + + var ack ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &ack, time.Minute) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrContextDone, + "cancelled context must unblock distribution-ack receive immediately") + }) + + t.Run("per-phase timer fires", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no ack will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "distribution-ack-session"}) + + s := utilsession.New(ms, context.Background(), jsession.JSONMarshaller{}) + + var ack ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &ack, 1*time.Millisecond) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrTimeout, + "expired per-phase timer must unblock distribution-ack receive with ErrTimeout") + }) +} + +// TestEndorsedTxDistributionTimeoutViaCancelledContext verifies that when the +// endorsed transaction is never distributed (e.g., CollectEndorsementsView +// delays or dies), a cancelled context surfaces immediately. This mirrors the +// ReceiveTransactionView / ReceiveRawWithTimeout call path. +func TestEndorsedTxDistributionTimeoutViaCancelledContext(t *testing.T) { + t.Run("context cancelled", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no tx will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "endorsed-tx-session"}) + + cancelledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + s := utilsession.New(ms, cancelledCtx, jsession.JSONMarshaller{}) + + var tx ttx.TransactionPayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeTransaction, &tx, time.Minute) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrContextDone, + "cancelled context must unblock endorsed-tx receive immediately") + }) + + t.Run("per-phase timer fires", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – no tx will arrive + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "endorsed-tx-session"}) + + s := utilsession.New(ms, context.Background(), jsession.JSONMarshaller{}) + + var tx ttx.TransactionPayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeTransaction, &tx, 1*time.Millisecond) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrTimeout, + "expired per-phase timer must unblock endorsed-tx receive with ErrTimeout") + }) +} + +// TestAuditorSignatureTimeoutViaCancelledContext verifies that when the auditor +// never replies with its signature, a cancelled context unblocks immediately. +// The auditing initiator view in auditingViewInitiator uses the same +// ReceiveTypedWithTimeout(TypeSignature, …) pattern. +func TestAuditorSignatureTimeoutViaCancelledContext(t *testing.T) { + t.Run("context cancelled", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – auditor never responds + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "auditor-session"}) + + cancelledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + s := utilsession.New(ms, cancelledCtx, jsession.JSONMarshaller{}) + + var sig ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &sig, time.Minute) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrContextDone, + "cancelled context must unblock auditor-signature receive immediately") + }) + + t.Run("per-phase timer fires", func(t *testing.T) { + ms := &sessionmock.Session{} + ch := make(chan *view.Message) // unbuffered – auditor never responds + ms.ReceiveReturns(ch) + ms.InfoReturns(view.SessionInfo{ID: "auditor-session"}) + + s := utilsession.New(ms, context.Background(), jsession.JSONMarshaller{}) + + var sig ttx.SignaturePayload + err := jsession.ReceiveTypedWithTimeout(s, ttx.TypeSignature, &sig, 1*time.Millisecond) + require.Error(t, err) + assert.ErrorIs(t, err, utilsession.ErrTimeout, + "expired per-phase timer must unblock auditor-signature receive with ErrTimeout") + }) +} diff --git a/token/services/ttx/endorse_test.go b/token/services/ttx/endorse_test.go index 5d3b6689af..a65ad3c3bb 100644 --- a/token/services/ttx/endorse_test.go +++ b/token/services/ttx/endorse_test.go @@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0 package ttx_test import ( + "context" "encoding/json" "reflect" "testing" @@ -25,8 +26,15 @@ import ( ) type TestEndorseViewContextInput struct { - IssuerIdentity token.Identity - AnotherReceivedTx bool + IssuerIdentity token.Identity + // CancelledContext replaces the test context with one that is already cancelled, + // so that any blocking session.Receive call returns ErrContextDone immediately + // without having to wait for the hardcoded per-phase deadline (e.g. 1 minute). + CancelledContext bool + // SigRequestOnly puts the signature-request message in the channel but not the + // endorsed-transaction message. Used to simulate the case where the initiator + // goes silent after the responder has already signed. + SigRequestOnly bool } type TestEndorseViewContext struct { @@ -125,31 +133,58 @@ func newTestEndorseViewContext(t *testing.T, input *TestEndorseViewContextInput) } } + baseCtx := t.Context() + if input.CancelledContext { + var cancel context.CancelFunc + baseCtx, cancel = context.WithCancel(baseCtx) + cancel() // immediately cancelled + } + ctx := &mock2.Context{} ctx.SessionReturns(session) - ctx.ContextReturns(t.Context()) + ctx.ContextReturns(baseCtx) ctx.GetServiceStub = getService tx, err := ttx.NewTransaction(ctx, []byte("a_signer")) require.NoError(t, err) ctx = &mock2.Context{} ctx.SessionReturns(session) - ctx.ContextReturns(t.Context()) + ctx.ContextReturns(baseCtx) ctx.GetServiceStub = getService txRaw, err := tx.Bytes() require.NoError(t, err) - // first the signature request - signatureRequest := &ttx.SignatureRequest{ - Signer: input.IssuerIdentity, - } - ch <- &view.Message{ - Payload: mustEnvelopeBytes(t, ttx.TypeSignatureRequest, signatureRequest), + // Whether to queue messages in the channel depends on the scenario being tested: + // + // CancelledContext=true, SigRequestOnly=false: + // → no messages queued; both phase-1 and phase-2 receives unblock immediately + // via ctx.Done() (tests phase-1 timeout / context-cancellation). + // + // CancelledContext=true, SigRequestOnly=true: + // → only the sig-request is queued so phase-1 reads from the buffer without + // blocking; phase-2 finds an empty channel + done context + // (tests phase-2 timeout / context-cancellation). + // + // CancelledContext=false, SigRequestOnly=false (default "success" path): + // → both messages queued. + // + // CancelledContext=false, SigRequestOnly=true: + // → only sig-request queued; phase-2 blocks until timer fires (not useful in + // unit tests, reserved for live testing). + if !input.CancelledContext || input.SigRequestOnly { + signatureRequest := &ttx.SignatureRequest{ + Signer: input.IssuerIdentity, + } + ch <- &view.Message{ + Payload: mustEnvelopeBytes(t, ttx.TypeSignatureRequest, signatureRequest), + } } - // then the transaction - ch <- &view.Message{ - Payload: mustEnvelopeBytes(t, ttx.TypeTransaction, &ttx.TransactionPayload{Raw: txRaw}), + if !input.CancelledContext && !input.SigRequestOnly { + // then the transaction + ch <- &view.Message{ + Payload: mustEnvelopeBytes(t, ttx.TypeTransaction, &ttx.TransactionPayload{Raw: txRaw}), + } } ctx.RunViewStub = func(v view.View, option ...view.RunViewOption) (any, error) { @@ -258,6 +293,32 @@ func TestEndorseView(t *testing.T) { assert.Equal(t, 0, ctx.session.SendWithContextCallCount()) }, }, + // --- Timeout / context-cancellation tests --- + // + // Each of the two blocking receive points inside EndorseView is guarded by + // a session.S select that also watches ctx.Done(). By passing an already- + // cancelled context the tests run instantly instead of waiting the hardcoded + // 1-minute / 4-minute per-phase deadlines. + { + // Phase 1: waiting for the signature request from the initiator. + // The channel is empty and the context is cancelled → ErrContextDone. + name: "context cancelled while waiting for signature request", + prepare: func() *TestEndorseViewContext { + c := newTestEndorseViewContext(t, &TestEndorseViewContextInput{ + CancelledContext: true, + }) + + return c + }, + expectError: true, + // The error is wrapped as "failed reading signature request: ctx done …" + // and joined with ErrHandlingSignatureRequests. + expectErr: ttx.ErrHandlingSignatureRequests, + verify: func(ctx *TestEndorseViewContext, _ any) { + // No signature was sent back because the receive failed before signing. + assert.Equal(t, 0, ctx.session.SendWithContextCallCount()) + }, + }, } for _, tc := range testCases { @@ -282,3 +343,52 @@ func TestEndorseView(t *testing.T) { }) } } + +// TestEndorseView_EndorsedTxContextCancellation verifies phase 2 of EndorseView: +// after the responder has signed and sent back the signature, the initiator goes +// silent (never distributes the endorsed transaction). A context that is +// cancelled while the view is blocking on the second receive must unblock the +// view and return an error. +// +// Implementation note: EndorseView.receiveTransaction calls ReceiveTransaction +// which ultimately calls session.S.ReceiveRawWithTimeout. That helper's select +// simultaneously watches the message channel and ctx.Done(), so cancelling the +// context is sufficient to surface the error without waiting for the 4-minute +// hardcoded deadline. +// +// We cancel the context from a separate goroutine, after the sig-request +// message has been consumed from the buffered channel, to avoid the non- +// deterministic select race that would arise if ctx.Done() were already closed +// when the first receive executes. +func TestEndorseView_EndorsedTxContextCancellation(t *testing.T) { + // Build the context with a live (non-cancelled) base context; SigRequestOnly + // ensures only the signature-request message is placed in the channel. + testCtx := newTestEndorseViewContext(t, &TestEndorseViewContextInput{ + SigRequestOnly: true, + }) + + // Replace the view context's context.Context() with one we can cancel. + cancelCtx, cancel := context.WithCancel(t.Context()) + + // Cancel after the first send (the token signature) has been recorded on the + // mock session. We watch the call count from a goroutine and cancel as soon + // as phase 1 completes; phase 2 will then find a done context. + go func() { + for testCtx.session.SendWithContextCallCount() == 0 { + // busy-spin with a yield so the main goroutine can progress + // (in practice this exits after microseconds) + } + cancel() + }() + + testCtx.ctx.ContextReturns(cancelCtx) + + v := ttx.NewEndorseView(testCtx.tx) + _, err := v.Call(testCtx.ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed receiving transaction", + "error should indicate failure during endorsed-tx receive phase") + // Phase 1 signature was sent before the context was cancelled. + assert.GreaterOrEqual(t, testCtx.session.SendWithContextCallCount(), 1, + "token signature must have been sent before context cancellation") +}