fix(selector): expire token locks by the consuming transaction

docs/imgs/storage_db.puml

@@ -127,7 +127,7 @@ package "Token Store (TokenDB)" {
         * idx : INT <<PK, NOT NULL, FK>>
         --
         consumer_tx_id : TEXT <<NOT NULL>>
-        created_at : TIMESTAMP <<NOT NULL>>
+        created_at : TIMESTAMPTZ <<NOT NULL>>
     }
 
     entity "TokenSKICleanups" as tkn_ski_cleanups {

docs/services/selector.md

@@ -119,6 +119,42 @@ To prevent double-spending *before* the transaction is committed to the ledger,
 2.  **Concurrency Control**: If another concurrent process has already locked that token, the insertion fails, and the selector moves on to the next candidate.
 3.  **Lock Release**: Locks are released either when the transaction reaches finality (success/failure) or when a timeout occurs, ensuring that tokens do not remain permanently inaccessible due to crashed or abandoned transactions.
 
+#### Lease expiry
+
+Every `leaseCleanupTickPeriod`, `sherdlock` runs a cleanup pass over the `TokenLocks`
+table that releases a lock when **either** of the following holds.
+
+> **Both `leaseExpiry` and `leaseCleanupTickPeriod` must be non-zero for the cleanup
+> goroutine to start.** If either is zero the pass never runs, so locks held by
+> `Deleted` or `Orphan` consumers are never released and those tokens remain
+> permanently unselectable. Setting `leaseExpiry: 0` to disable time-based expiry
+> while relying on consumer-status release is therefore not supported.
+
+*   the **consuming** transaction — the one that took the lock, stored in
+    `consumer_tx_id` — has reached `Deleted` or `Orphan`, so it will never spend the
+    token; or
+*   the lease is older than `leaseExpiry`, which covers the consumer that crashed or was
+    abandoned without ever reaching a terminal status.
+
+Two properties of the pass are worth spelling out:
+
+*   **The status that matters is the consumer's, not the producer's.** A lock row is keyed
+    by `(tx_id, idx)`, which identifies the *locked token* and therefore the transaction
+    that created it. That transaction's status says nothing about whether the lock is
+    still live, so it is never used to expire a lease.
+*   **Expiry is per token, not per transaction.** Only the affected `(tx_id, idx)` rows are
+    deleted; the other outputs of the same transaction keep their locks.
+
+The pass is the same statement on every SQL backend (SQLite and Postgres), so lock
+expiry behaviour is identical across those backends. `created_at` is stored as
+`TIMESTAMPTZ`, so the comparison with the database-side `NOW()` expression is always
+timezone-consistent on Postgres regardless of the session `TimeZone` setting.
+On Postgres a single replica per TMS runs the pass per tick, elected through an advisory
+lock; SQLite is non-distributed and always runs it locally.
+
+The in-memory locker described below does not use the `TokenLocks` table and never
+expires locks via `Cleanup`; its lifecycle is entirely managed in process.
+
 ### In-Memory Locker Internals
 
 The `simple` driver keeps its locks in memory (`token/services/selector/simple/inmemory`)

token/services/storage/db/dbtest/tokenlock.go

@@ -14,9 +14,13 @@ import (
 	driver3 "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver"
 	"github.com/LFDT-Panurus/panurus/token/services/utils"
 	"github.com/LFDT-Panurus/panurus/token/token"
+	fscerrors "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/stretchr/testify/require"
 )
 
+// agedLease is the leaseExpiry used for the aged-lease test.
+const agedLease = time.Hour
+
 func TokenLocksTest(t *testing.T, cfgProvider cfgProvider) {
 	t.Helper()
 	for _, c := range tokenLockDBCases {
@@ -55,6 +59,12 @@ var tokenLockDBCases = []struct {
 	Fn   func(*testing.T, driver3.TokenStore, driver3.TokenLockStore, driver3.TokenTransactionStore)
 }{
 	{"TestFully", TestFully},
+	{"TestReleaseOnDeletedConsumer", TestReleaseOnDeletedConsumer},
+	{"TestReleaseOnOrphanConsumer", TestReleaseOnOrphanConsumer},
+	{"TestKeepOnDeletedProducer", TestKeepOnDeletedProducer},
+	{"TestKeepSiblingIndices", TestKeepSiblingIndices},
+	{"TestReleaseOnAgedLease", TestReleaseOnAgedLease},
+	{"TestKeepFreshPendingLock", TestKeepFreshPendingLock},
 }
 
 func TestFully(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
@@ -97,3 +107,177 @@ func TestFully(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.Tok
 	// Cleanup should work correctly
 	require.NoError(t, tokenLockDB.Cleanup(ctx, 1*time.Second))
 }
+
+// longLease outlives any of these tests, so a Cleanup call using it can only collect
+// locks through the status of their consuming transaction, never through lease ageing.
+const longLease = time.Hour
+
+// TestReleaseOnDeletedConsumer verifies that a lease is released as soon as the
+// transaction that was going to spend the token is Deleted, rather than being held
+// until the lease ages out. See #2018.
+func TestReleaseOnDeletedConsumer(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	tokenID := token.ID{TxId: "producer", Index: 0}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "consumer")
+	storeTokens(t, tokenDB, "producer", 0)
+	require.NoError(t, tokenLockDB.Lock(ctx, &tokenID, "consumer", "owner1"))
+
+	require.NoError(t, tokenTransactionDB.SetStatus(ctx, "consumer", driver3.Deleted, ""))
+	require.NoError(t, tokenLockDB.Cleanup(ctx, longLease))
+
+	requireLockReleased(t, tokenLockDB, tokenID)
+}
+
+// TestReleaseOnOrphanConsumer verifies that an Orphan consuming transaction releases
+// its leases too, on the same tick as a Deleted one. See #2018.
+func TestReleaseOnOrphanConsumer(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	tokenID := token.ID{TxId: "producer", Index: 0}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "consumer")
+	storeTokens(t, tokenDB, "producer", 0)
+	require.NoError(t, tokenLockDB.Lock(ctx, &tokenID, "consumer", "owner1"))
+
+	require.NoError(t, tokenTransactionDB.SetStatus(ctx, "consumer", driver3.Orphan, ""))
+	require.NoError(t, tokenLockDB.Cleanup(ctx, longLease))
+
+	requireLockReleased(t, tokenLockDB, tokenID)
+}
+
+// TestKeepOnDeletedProducer verifies that the status of the transaction that created
+// the locked token does not expire the lease: the consuming transaction is still in
+// flight, so dropping its lock would let the token be selected twice. See #2018.
+func TestKeepOnDeletedProducer(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	tokenID := token.ID{TxId: "producer", Index: 0}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "consumer")
+	storeTokens(t, tokenDB, "producer", 0)
+	require.NoError(t, tokenLockDB.Lock(ctx, &tokenID, "consumer", "owner1"))
+
+	require.NoError(t, tokenTransactionDB.SetStatus(ctx, "producer", driver3.Deleted, ""))
+	require.NoError(t, tokenLockDB.Cleanup(ctx, longLease))
+
+	requireLockHeld(t, tokenLockDB, tokenID)
+}
+
+// TestKeepSiblingIndices verifies that expiring the lock of one output does not take
+// the locks of the other outputs of the same transaction with it: the primary key of
+// the lock table is (tx_id, idx), so cleanup must be scoped to both. See #2018.
+func TestKeepSiblingIndices(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	expired := token.ID{TxId: "producer", Index: 0}
+	live := token.ID{TxId: "producer", Index: 1}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "dead-consumer")
+	addTokenRequest(t, tokenTransactionDB, "live-consumer")
+	storeTokens(t, tokenDB, "producer", 0, 1)
+	require.NoError(t, tokenLockDB.Lock(ctx, &expired, "dead-consumer", "owner1"))
+	require.NoError(t, tokenLockDB.Lock(ctx, &live, "live-consumer", "owner1"))
+
+	require.NoError(t, tokenTransactionDB.SetStatus(ctx, "dead-consumer", driver3.Deleted, ""))
+	require.NoError(t, tokenLockDB.Cleanup(ctx, longLease))
+
+	requireLockHeld(t, tokenLockDB, live)
+	requireLockReleased(t, tokenLockDB, expired)
+}
+
+// TestReleaseOnAgedLease verifies the second expiry branch: a lock whose consuming
+// transaction never reaches a terminal status is reclaimed once its lease is older
+// than leaseExpiry.
+//
+// The lock is inserted with a backdated created_at so no sleep is needed.
+// The threshold used in Cleanup (agedLease = 1h) is safely larger than any
+// clock skew between the test process and the database.
+func TestReleaseOnAgedLease(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	tokenID := token.ID{TxId: "producer", Index: 0}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "consumer")
+	storeTokens(t, tokenDB, "producer", 0)
+	require.NoError(t, tokenLockDB.LockAt(ctx, &tokenID, "consumer", "owner1", time.Now().Add(-2*agedLease)))
+	require.NoError(t, tokenLockDB.Cleanup(ctx, agedLease))
+
+	requireLockReleased(t, tokenLockDB, tokenID)
+}
+
+// TestKeepFreshPendingLock verifies that cleanup leaves alone a fresh lock whose
+// consuming transaction is still pending - neither expiry branch applies to it.
+func TestKeepFreshPendingLock(t *testing.T, tokenDB driver3.TokenStore, tokenLockDB driver3.TokenLockStore, tokenTransactionDB driver3.TokenTransactionStore) {
+	ctx := t.Context()
+	tokenID := token.ID{TxId: "producer", Index: 0}
+
+	addTokenRequest(t, tokenTransactionDB, "producer")
+	addTokenRequest(t, tokenTransactionDB, "consumer")
+	storeTokens(t, tokenDB, "producer", 0)
+	require.NoError(t, tokenLockDB.Lock(ctx, &tokenID, "consumer", "owner1"))
+
+	require.NoError(t, tokenLockDB.Cleanup(ctx, longLease))
+
+	requireLockHeld(t, tokenLockDB, tokenID)
+}
+
+// addTokenRequest registers a token request for txID, so that its status can later be
+// moved to a terminal one with SetStatus.
+func addTokenRequest(t *testing.T, tokenTransactionDB driver3.TokenTransactionStore, txID string) {
+	t.Helper()
+
+	tx, err := tokenTransactionDB.NewTransactionStoreTransaction()
+	require.NoError(t, err)
+	require.NoError(t, tx.AddTokenRequest(t.Context(), txID, []byte(txID+"_tx_content"), nil, nil, driver2.PPHash("tr")))
+	require.NoError(t, tx.Commit())
+}
+
+// storeTokens stores one owned token per index of txID, so that the (tx_id, idx)
+// foreign key carried by the lock rows is satisfied.
+func storeTokens(t *testing.T, tokenDB driver3.TokenStore, txID string, indices ...uint64) {
+	t.Helper()
+
+	tx, err := tokenDB.NewTokenDBTransaction()
+	require.NoError(t, err)
+	for _, index := range indices {
+		require.NoError(t, tx.StoreToken(t.Context(), driver3.TokenRecord{
+			TxID:           txID,
+			Index:          index,
+			OwnerRaw:       []byte("owner1"),
+			OwnerType:      "idemix",
+			OwnerIdentity:  []byte("owner1"),
+			Ledger:         []byte("ledger_data"),
+			LedgerMetadata: []byte{},
+			Quantity:       "0x64",
+			Type:           "USD",
+			Amount:         100,
+			Owner:          true,
+		}, []string{"owner1"}))
+	}
+	require.NoError(t, tx.Commit())
+}
+
+// requireLockHeld asserts that the lock on tokenID survived cleanup. The store exposes
+// no read API, so the probe is a second Lock on the same token: the (tx_id, idx)
+// primary key rejects it for as long as the row is there.
+// We assert on driver3.ErrTokenAlreadyLocked specifically so that an unrelated Lock
+// failure (e.g. a future rule that rejects locks on Deleted producers) does not make
+// "Keep" tests pass vacuously.
+func requireLockHeld(t *testing.T, tokenLockDB driver3.TokenLockStore, tokenID token.ID) {
+	t.Helper()
+
+	err := tokenLockDB.Lock(t.Context(), &tokenID, "probe-"+tokenID.String(), "owner1")
+	require.True(t, fscerrors.Is(err, driver3.ErrTokenAlreadyLocked),
+		"lock on token %s should still be held (want ErrTokenAlreadyLocked, got %v)", tokenID, err)
+}
+
+// requireLockReleased asserts that cleanup collected the lock on tokenID: the row is
+// gone, so the token can be locked again.
+func requireLockReleased(t *testing.T, tokenLockDB driver3.TokenLockStore, tokenID token.ID) {
+	t.Helper()
+
+	require.NoError(t, tokenLockDB.Lock(t.Context(), &tokenID, "probe-"+tokenID.String(), "owner1"),
+		"lock on token %s should have been released", tokenID)
+}

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

@@ -336,12 +336,22 @@ type TokenLockStore interface {
 	// the tokens are selected for). The built-in SQL store ignores walletID; a custom
 	// store may use it to apply per-wallet policies such as rate limiting, returning an
 	// error wrapping token.SelectorRateLimited to make the selection fail fast.
+	// The lock timestamp is set to the current time.
 	Lock(ctx context.Context, tokenID *token.ID, consumerTxID transaction.ID, walletID string) error
+	// LockAt is like Lock but records the supplied timestamp as the lock creation
+	// time instead of the current time. It is intended for testing (backdating locks
+	// to exercise the lease-age expiry path without sleeping).
+	LockAt(ctx context.Context, tokenID *token.ID, consumerTxID transaction.ID, walletID string, createdAt time.Time) error
 	// UnlockByTxID unlocks all tokens locked by the consumer TX
 	UnlockByTxID(ctx context.Context, consumerTxID transaction.ID) error
-	// Cleanup removes the locks such that either:
-	// 1. The transaction that locked that token is valid or invalid;
-	// 2. The lock is too old.
+	// Cleanup removes stale token locks. A lock is stale when either:
+	// 1. The *consuming* transaction (the one trying to spend the token,
+	//    identified by consumer_tx_id) has reached a terminal failure status
+	//    (Deleted or Orphan); or
+	// 2. The lock is older than leaseExpiry (covers consumers that crashed
+	//    before reaching any terminal status).
+	// The producer transaction (the one that created the token, identified by
+	// (tx_id, idx)) is irrelevant to expiry; see #2018.
 	Cleanup(ctx context.Context, leaseExpiry time.Duration) error
 	// AcquireCleanupLeadership attempts to acquire leadership for the
 	// cleanup tick, so only one replica runs Cleanup per tick. Non-distributed
@@ -353,4 +363,9 @@ type TokenLockStore interface {
 	Close() error
 }
 
-var ErrTokenDoesNotExist = errors.New("token does not exist")
+var (
+	ErrTokenDoesNotExist   = errors.New("token does not exist")
+	// ErrTokenAlreadyLocked is returned by TokenLockStore.Lock when the token is
+	// already locked by another transaction (primary-key conflict on the lock row).
+	ErrTokenAlreadyLocked  = errors.New("token already locked")
+)