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

docs/services/selector.md

@@ -119,6 +119,31 @@ 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:
+
+*   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, so lock lifetime does not depend on
+whether a deployment runs on SQLite, the in-memory backend, or Postgres. On Postgres a
+single replica per TMS runs it per tick, elected through an advisory lock; SQLite is
+non-distributed and always runs it locally.
+
 ### In-Memory Locker Internals
 
 The `simple` driver keeps its locks in memory (`token/services/selector/simple/inmemory`)

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

@@ -55,6 +55,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 +103,174 @@ 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.
+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.Lock(ctx, &tokenID, "consumer", "owner1"))
+
+	// The margin over the lease is deliberate: the SQLite interpreter renders the
+	// threshold with datetime('now', '-N seconds'), which has one-second resolution,
+	// so a sub-second margin would make this flaky.
+	time.Sleep(2200 * time.Millisecond)
+	require.NoError(t, tokenLockDB.Cleanup(ctx, time.Second))
+
+	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.
+func requireLockHeld(t *testing.T, tokenLockDB driver3.TokenLockStore, tokenID token.ID) {
+	t.Helper()
+
+	require.Error(t, tokenLockDB.Lock(t.Context(), &tokenID, "probe-"+tokenID.String(), "owner1"),
+		"lock on token %s should still be held", tokenID)
+}
+
+// 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/sql/common/tokenlock.go

@@ -117,3 +117,45 @@ func IsExpiredToken(tokenRequests, tokenLocks common3.Table, leaseExpiry time.Du
 		cond.OlderThan(tokenLocks.Field("created_at"), leaseExpiry),
 	)
 }
+
+// IsStaleLock matches the lock rows whose lease has aged out, or whose consuming
+// transaction is Deleted or Orphan. The correlation is on consumer_tx_id, the
+// transaction that is trying to spend the token: (tx_id, idx) identifies the locked
+// token, i.e. the transaction that created it, whose status says nothing about
+// whether the lock is still live. The condition is correlated rather than a
+// partial-key IN on tx_id, so cleanup removes only the matching (tx_id, idx) rows
+// and leaves the other indices of the same transaction locked. See #2018.
+func IsStaleLock(tokenLocks, tokenRequests common3.Table, leaseExpiry time.Duration) cond.Condition {
+	return cond.Or(
+		cond.OlderThan(tokenLocks.Field("created_at"), leaseExpiry),
+		cond.Exists(
+			q.Select().
+				Fields(common3.FieldName("1")).
+				From(tokenRequests).
+				Where(cond.And(
+					cond.Cmp(tokenRequests.Field("tx_id"), "=", tokenLocks.Field("consumer_tx_id")),
+					cond.FieldIn(tokenRequests.Field("status"), driver.Deleted, driver.Orphan),
+				)),
+		),
+	)
+}
+
+// Cleanup releases the stale token locks: those whose consuming transaction is
+// Deleted or Orphan, and those whose lease is older than leaseExpiry. Only the
+// affected (tx_id, idx) rows are deleted. The same statement is used by every SQL
+// backend, so lock lifetime does not depend on the driver in use.
+func (db *TokenLockStore) Cleanup(ctx context.Context, leaseExpiry time.Duration) error {
+	tokenLocks, tokenRequests := q.Table(db.Table.TokenLocks), q.Table(db.Table.Requests)
+
+	query, args := q.DeleteFrom(db.Table.TokenLocks).
+		Where(IsStaleLock(tokenLocks, tokenRequests, leaseExpiry)).
+		Format(db.ci)
+
+	db.Logger.Debug(query, args)
+	_, err := db.WriteDB.ExecContext(ctx, query, args...)
+	if err != nil {
+		db.Logger.Errorf("query failed: %s", query)
+	}
+
+	return err
+}

token/services/storage/db/sql/postgres/tokenlock.go

@@ -85,37 +85,15 @@ func (db *TokenLockStore) AcquireCleanupLeadership(ctx context.Context) (driver.
 	return db.cleanupLeaderFactory(ctx, db.writeDB, db.cleanupLockID)
 }
 
-// Cleanup removes stale token locks that have expired.
+// Cleanup removes stale token locks that have expired. The deletion itself is the
+// backend-independent one implemented by the embedded store; Postgres only adds the
+// logging of the rows that are about to go.
 func (db *TokenLockStore) Cleanup(ctx context.Context, leaseExpiry time.Duration) error {
 	if err := db.logStaleLocks(ctx, leaseExpiry); err != nil {
 		db.Logger.Warnf("Could not log stale locks: %v", err)
 	}
-	tokenLocks, tokenRequests := q.Table(db.Table.TokenLocks), q.Table(db.Table.Requests)
-
-	existsDeletedOrOrphan := cond.Exists(
-		q.Select().
-			Fields(common3.FieldName("1")).
-			From(tokenRequests).
-			Where(cond.And(
-				cond.Cmp(tokenRequests.Field("tx_id"), "=", tokenLocks.Field("consumer_tx_id")),
-				cond.FieldIn(tokenRequests.Field("status"), driver.Deleted, driver.Orphan),
-			)),
-	)
-
-	query, args := q.DeleteFrom(db.Table.TokenLocks).
-		Where(cond.Or(
-			cond.OlderThan(tokenLocks.Field("created_at"), leaseExpiry),
-			existsDeletedOrOrphan,
-		)).
-		Format(db.ci)
-
-	db.Logger.Debug(query)
-	_, err := db.WriteDB.ExecContext(ctx, query, args...)
-	if err != nil {
-		db.Logger.Errorf("query failed: %s", query)
-	}
 
-	return err
+	return db.TokenLockStore.Cleanup(ctx, leaseExpiry)
 }
 
 // logStaleLocks logs the token locks that are about to be deleted.

token/services/storage/db/sql/sqlite/tokenlock.go

@@ -8,56 +8,17 @@ package sqlite
 
 import (
 	"context"
-	"time"
 
-	q "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query"
-	common2 "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/common"
-	"github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/cond"
 	common3 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/common"
 
 	"github.com/LFDT-Panurus/panurus/token/services/storage/db/driver"
 	common4 "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/common"
 )
 
+// TokenLockStore implements the token lock storage for SQLite. Cleanup is inherited
+// from the embedded store, so lease expiry is identical to the other SQL backends.
 type TokenLockStore struct {
 	*common4.TokenLockStore
-	ci common2.CondInterpreter
-}
-
-func IsStale(tokenLocks common2.TableName, requests common2.TableName, leaseExpiry time.Duration) *isStale {
-	return &isStale{tokenLocks: tokenLocks, requests: requests, leaseExpiry: leaseExpiry}
-}
-
-type isStale struct {
-	tokenLocks  common2.TableName
-	requests    common2.TableName
-	leaseExpiry time.Duration
-}
-
-func (c *isStale) WriteString(ci common2.CondInterpreter, sb common2.Builder) {
-	tokenLocks, tokenRequests := q.AliasedTable(string(c.tokenLocks), "tl"), q.AliasedTable(string(c.requests), "tr")
-
-	sb.WriteString("tx_id IN (")
-	q.Select().
-		Fields(tokenLocks.Field("tx_id")).
-		From(tokenLocks.Join(tokenRequests, cond.Cmp(tokenLocks.Field("tx_id"), "=", tokenRequests.Field("tx_id")))).
-		Where(common4.IsExpiredToken(tokenRequests, tokenLocks, c.leaseExpiry)).
-		FormatTo(ci, sb)
-	sb.WriteRune(')')
-}
-
-func (db *TokenLockStore) Cleanup(ctx context.Context, leaseExpiry time.Duration) error {
-	query, args := q.DeleteFrom(db.Table.TokenLocks).
-		Where(IsStale(common2.TableName(db.Table.TokenLocks), common2.TableName(db.Table.Requests), leaseExpiry)).
-		Format(db.ci)
-
-	db.Logger.Debug(query, args)
-	_, err := db.WriteDB.ExecContext(ctx, query, args...)
-	if err != nil {
-		db.Logger.Errorf("query failed: %s", query)
-	}
-
-	return err
 }
 
 // AcquireCleanupLeadership always grants leadership locally - sqlite is a
@@ -67,11 +28,12 @@ func (db *TokenLockStore) AcquireCleanupLeadership(_ context.Context) (driver.Cl
 	return driver.NoopCleanupLeadership{}, true, nil
 }
 
+// NewTokenLockStore returns a new TokenLockStore for the given RWDB and table names.
 func NewTokenLockStore(dbs *common3.RWDB, tableNames common4.TableNames) (*TokenLockStore, error) {
 	tldb, err := common4.NewTokenLockStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter())
 	if err != nil {
 		return nil, err
 	}
 
-	return &TokenLockStore{TokenLockStore: tldb, ci: NewConditionInterpreter()}, nil
+	return &TokenLockStore{TokenLockStore: tldb}, nil
 }