enforce database timeouts

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)
+}

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()))
 

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)
+}

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)

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

@@ -7,13 +7,15 @@ SPDX-License-Identifier: Apache-2.0
 package common
 
 import (
+	"context"
 	"database/sql"
 	driver2 "database/sql/driver"
 	"math/big"
 	"testing"
 	"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)
+}

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)
+}

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)
+}