add an option to stop the htlc scan for preimage when all blocks are … (

hyperledger-labs#627)

Signed-off-by: Arne Rutjes <[email protected]>

integration/token/interop/support.go

@@ -390,7 +390,7 @@ func fastExchange(network *integration.Infrastructure, id string, recipient stri
 	time.Sleep(10 * time.Second)
 }
 
-func scan(network *integration.Infrastructure, id string, hash []byte, hashFunc crypto.Hash, startingTransactionID string, opts ...token.ServiceOption) {
+func scan(network *integration.Infrastructure, id string, hash []byte, hashFunc crypto.Hash, startingTransactionID string, stopOnLastTx bool, opts ...token.ServiceOption) {
 	options, err := token.CompileServiceOptions(opts...)
 	Expect(err).NotTo(HaveOccurred())
 
@@ -400,11 +400,12 @@ func scan(network *integration.Infrastructure, id string, hash []byte, hashFunc
 		Hash:                  hash,
 		HashFunc:              hashFunc,
 		StartingTransactionID: startingTransactionID,
+		StopOnLastTx:          stopOnLastTx,
 	}))
 	Expect(err).NotTo(HaveOccurred())
 }
 
-func scanWithError(network *integration.Infrastructure, id string, hash []byte, hashFunc crypto.Hash, startingTransactionID string, errorMsgs []string, opts ...token.ServiceOption) {
+func scanWithError(network *integration.Infrastructure, id string, hash []byte, hashFunc crypto.Hash, startingTransactionID string, errorMsgs []string, stopOnLastTx bool, opts ...token.ServiceOption) {
 	options, err := token.CompileServiceOptions(opts...)
 	Expect(err).NotTo(HaveOccurred())
 
@@ -414,6 +415,7 @@ func scanWithError(network *integration.Infrastructure, id string, hash []byte,
 		Hash:                  hash,
 		HashFunc:              hashFunc,
 		StartingTransactionID: startingTransactionID,
+		StopOnLastTx:          stopOnLastTx,
 	}))
 	Expect(err).To(HaveOccurred())
 	for _, msg := range errorMsgs {

integration/token/interop/tests.go

@@ -258,19 +258,24 @@ func TestHTLCNoCrossClaimTwoNetworks(network *integration.Infrastructure) {
 
 	go func() { htlcClaim(network, alpha, "alice", "alice.id2", preImage) }()
 	go func() { htlcClaim(network, beta, "bob", "bob.id2", preImage) }()
-	scan(network, "alice", hash, crypto.SHA256, "", token.WithTMSID(alpha))
-	scan(network, "alice", hash, crypto.SHA256, aliceLockTxID, token.WithTMSID(alpha))
+	scan(network, "alice", hash, crypto.SHA256, "", false, token.WithTMSID(alpha))
+	scan(network, "alice", hash, crypto.SHA256, aliceLockTxID, false, token.WithTMSID(alpha))
 
-	scan(network, "bob", hash, crypto.SHA256, "", token.WithTMSID(beta))
-	scan(network, "bob", hash, crypto.SHA256, bobLockTxID, token.WithTMSID(beta))
+	scan(network, "bob", hash, crypto.SHA256, "", false, token.WithTMSID(beta))
+	scan(network, "bob", hash, crypto.SHA256, bobLockTxID, false, token.WithTMSID(beta))
 
 	CheckBalanceWithLockedAndHolding(network, "alice", "alice.id1", "EUR", 20, 0, 0, -1, token.WithTMSID(alpha))
 	CheckBalanceWithLockedAndHolding(network, "alice", "alice.id2", "EUR", 10, 0, 0, -1, token.WithTMSID(alpha))
 	CheckBalanceWithLockedAndHolding(network, "bob", "bob.id1", "USD", 20, 0, 0, -1, token.WithTMSID(beta))
 	CheckBalanceWithLockedAndHolding(network, "bob", "bob.id2", "USD", 10, 0, 0, -1, token.WithTMSID(beta))
 
 	txID := IssueCashWithTMS(network, alpha, "issuer", "", "EUR", 30, "alice.id1")
-	scanWithError(network, "alice", hash, crypto.SHA256, txID, []string{"timeout reached"}, token.WithTMSID(alpha))
+
+	scan(network, "bob", hash, crypto.SHA256, bobLockTxID, true, token.WithTMSID(beta))
+	start := time.Now()
+	scanWithError(network, "alice", hash, crypto.SHA256, txID, []string{"context done"}, true, token.WithTMSID(alpha))
+	Expect(time.Since(start)).To(BeNumerically("<", time.Second*30), "scan should be canceled on last tx, before timeout")
+	scanWithError(network, "alice", hash, crypto.SHA256, txID, []string{"timeout reached"}, false, token.WithTMSID(alpha))
 
 	CheckPublicParams(network, token.TMSID{}, "alice", "bob")
 	CheckPublicParams(network, alpha, "issuer", "auditor")

integration/token/interop/views/htlc/scan.go

@@ -31,22 +31,30 @@ type Scan struct {
 	// StartingTransactionID  is the transaction id from which to start the scan.
 	// If empty, the scan starts from the genesis block
 	StartingTransactionID string
+	// StopOnLastTx stops the scan if the last transaction is reached.
+	StopOnLastTx bool
 }
 
 type ScanView struct {
 	*Scan
 }
 
 func (s *ScanView) Call(context view.Context) (interface{}, error) {
+	opts := []token.ServiceOption{
+		token.WithTMSID(s.TMSID),
+		htlc.WithStartingTransaction(s.StartingTransactionID),
+	}
+	if s.StopOnLastTx {
+		opts = append(opts, htlc.WithStopOnLastTransaction())
+	}
+
 	preImage, err := htlc.ScanForPreImage(
 		context,
 		s.Hash,
 		s.HashFunc,
 		encoding.None,
 		s.Timeout,
-		token.WithTMSID(s.TMSID),
-		htlc.WithStartingTransaction(s.StartingTransactionID),
-	)
+		opts...)
 	assert.NoError(err, "failed to scan for pre-image")
 	return preImage, nil
 }

token/services/interop/htlc/scanner.go

@@ -20,6 +20,8 @@ import (
 
 const (
 	ScanForPreImageStartingTransaction = "htlc.ScanForPreImage.StartingTransaction"
+	StopScanningOnLastTransaction      = "htlc.ScanForPreImage.StopOnLastTransaction"
+	True                               = "true"
 )
 
 // WithStartingTransaction sets the network name
@@ -33,6 +35,18 @@ func WithStartingTransaction(txID string) token.ServiceOption {
 	}
 }
 
+// WithStopOnLastTransaction stops the scan when the last transaction is reached.
+// When this is not set, the scan will wait until timeout or until the key is found.
+func WithStopOnLastTransaction() token.ServiceOption {
+	return func(o *token.ServiceOptions) error {
+		if o.Params == nil {
+			o.Params = map[string]interface{}{}
+		}
+		o.Params[StopScanningOnLastTransaction] = True
+		return nil
+	}
+}
+
 // ScanForPreImage scans the ledger for a preimage of the passed image, taking into account the timeout
 func ScanForPreImage(sp token.ServiceProvider, image []byte, hashFunc crypto.Hash, hashEncoding encoding.Encoding, timeout time.Duration, opts ...token.ServiceOption) ([]byte, error) {
 	logger.Debugf("scanning for preimage of [%s] with timeout [%s]", base64.StdEncoding.EncodeToString(image), timeout)
@@ -59,9 +73,15 @@ func ScanForPreImage(sp token.ServiceProvider, image []byte, hashFunc crypto.Has
 	if err != nil {
 		return nil, errors.Wrapf(err, "invalid starting transaction param")
 	}
+	var stopOnLastTx bool
+	stop, err := tokenOptions.ParamAsString(StopScanningOnLastTransaction)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid stop on last transaction param")
+	}
+	stopOnLastTx = stop == True
 
 	claimKey := ClaimKey(image)
-	preImage, err := network.LookupTransferMetadataKey(tms.Namespace(), startingTxID, claimKey, timeout, opts...)
+	preImage, err := network.LookupTransferMetadataKey(tms.Namespace(), startingTxID, claimKey, timeout, stopOnLastTx, opts...)
 	if err != nil {
 		return nil, errors.WithMessagef(err, "failed to lookup key [%s]", claimKey)
 	}

token/services/network/driver/network.go

@@ -82,8 +82,8 @@ type Network interface {
 	RemoveFinalityListener(id string, listener FinalityListener) error
 
 	// LookupTransferMetadataKey searches for a transfer metadata key containing the passed sub-key starting from the passed transaction id in the given namespace.
-	// The operation gets canceled if the passed timeout elapses.
-	LookupTransferMetadataKey(namespace string, startingTxID string, subKey string, timeout time.Duration) ([]byte, error)
+	// The operation gets canceled if the passed timeout elapses or, if stopOnLastTx is true, when the last transaction in the vault is reached.
+	LookupTransferMetadataKey(namespace string, startingTxID string, subKey string, timeout time.Duration, stopOnLastTx bool) ([]byte, error)
 
 	// Ledger gives access to the remote ledger
 	Ledger() (Ledger, error)

token/services/network/fabric/network.go

@@ -338,7 +338,7 @@ func (n *Network) RemoveFinalityListener(txID string, listener driver.FinalityLi
 	return n.ch.Committer().RemoveFinalityListener(txID, wrapper.(*FinalityListener))
 }
 
-func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID string, key string, timeout time.Duration) ([]byte, error) {
+func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID string, key string, timeout time.Duration, stopOnLastTx bool) ([]byte, error) {
 	transferMetadataKey, err := keys.CreateTransferActionMetadataKey(key)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to generate transfer action metadata key from [%s]", key)
@@ -347,6 +347,16 @@ func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID strin
 	c, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	v := n.ch.Vault()
+
+	var lastTxID string
+	if stopOnLastTx {
+		id, err := v.GetLastTxID()
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get last transaction id")
+		}
+		lastTxID = id
+	}
+
 	if err := n.ch.Delivery().Scan(c, startingTxID, func(tx *fabric.ProcessedTransaction) (bool, error) {
 		logger.Debugf("scanning [%s]...", tx.TxID())
 
@@ -379,6 +389,11 @@ func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID strin
 			}
 		}
 		logger.Debugf("scanning for key [%s] on [%s] not found", transferMetadataKey, tx.TxID())
+		if stopOnLastTx && lastTxID == tx.TxID() {
+			logger.Debugf("final transaction reached on [%s]", tx.TxID())
+			cancel()
+		}
+
 		return false, nil
 	}); err != nil {
 		if strings.Contains(err.Error(), "context done") {

token/services/network/network.go

@@ -372,9 +372,9 @@ func (n *Network) RemoveFinalityListener(id string, listener FinalityListener) e
 }
 
 // LookupTransferMetadataKey searches for a transfer metadata key containing the passed sub-key starting from the passed transaction id in the given namespace.
-// The operation gets canceled if the passed timeout gets reached.
-func (n *Network) LookupTransferMetadataKey(namespace, startingTxID, key string, timeout time.Duration, opts ...token.ServiceOption) ([]byte, error) {
-	return n.n.LookupTransferMetadataKey(namespace, startingTxID, key, timeout)
+// The operation gets canceled if the passed timeout gets reached or, if stopOnLastTx is true, when the last transaction in the vault is reached.
+func (n *Network) LookupTransferMetadataKey(namespace, startingTxID, key string, timeout time.Duration, stopOnLastTx bool, opts ...token.ServiceOption) ([]byte, error) {
+	return n.n.LookupTransferMetadataKey(namespace, startingTxID, key, timeout, stopOnLastTx)
 }
 
 func (n *Network) Ledger() (*Ledger, error) {

token/services/network/orion/network.go

@@ -225,7 +225,7 @@ func (n *Network) RemoveFinalityListener(txID string, listener driver.FinalityLi
 	return n.n.Committer().RemoveFinalityListener(txID, wrapper.(*FinalityListener))
 }
 
-func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID string, key string, timeout time.Duration) ([]byte, error) {
+func (n *Network) LookupTransferMetadataKey(namespace string, startingTxID string, key string, timeout time.Duration, _ bool) ([]byte, error) {
 	k, err := keys.CreateTransferActionMetadataKey(key)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to generate transfer action metadata key from [%s]", key)