diff --git a/docs/imgs/storage_db.puml b/docs/imgs/storage_db.puml index db54ca788f..afc19ecefc 100644 --- a/docs/imgs/storage_db.puml +++ b/docs/imgs/storage_db.puml @@ -127,7 +127,7 @@ package "Token Store (TokenDB)" { * idx : INT <> -- consumer_tx_id : TEXT <> - created_at : TIMESTAMP <> + created_at : TIMESTAMPTZ <> } entity "TokenSKICleanups" as tkn_ski_cleanups { diff --git a/docs/services/selector.md b/docs/services/selector.md index 3cb53ccf4f..4ea50bd992 100644 --- a/docs/services/selector.md +++ b/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`) diff --git a/token/services/storage/db/dbtest/tokenlock.go b/token/services/storage/db/dbtest/tokenlock.go index 94c8ada5fd..c08c08e95e 100644 --- a/token/services/storage/db/dbtest/tokenlock.go +++ b/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) +} diff --git a/token/services/storage/db/driver/token.go b/token/services/storage/db/driver/token.go index 40058789df..63ebf97621 100644 --- a/token/services/storage/db/driver/token.go +++ b/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") +) diff --git a/token/services/storage/db/sql/common/tokenlock.go b/token/services/storage/db/sql/common/tokenlock.go index da4b5c498e..75b829d562 100644 --- a/token/services/storage/db/sql/common/tokenlock.go +++ b/token/services/storage/db/sql/common/tokenlock.go @@ -19,7 +19,9 @@ import ( "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/cond" "github.com/LFDT-Panurus/panurus/token/services/utils/types/transaction" "github.com/LFDT-Panurus/panurus/token/token" + "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" common2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/common" + fscdriver "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver" "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/common" ) @@ -30,24 +32,26 @@ type tokenLockTables struct { } type TokenLockStore struct { - ReadDB *sql.DB - WriteDB *sql.DB - Table tokenLockTables - Logger logging.Logger - ci common3.CondInterpreter + ReadDB *sql.DB + WriteDB *sql.DB + Table tokenLockTables + Logger logging.Logger + ci common3.CondInterpreter + errorWrapper fscdriver.SQLErrorWrapper } -func newTokenLockStore(readDB, writeDB *sql.DB, tables tokenLockTables, ci common3.CondInterpreter) *TokenLockStore { +func newTokenLockStore(readDB, writeDB *sql.DB, tables tokenLockTables, ci common3.CondInterpreter, errorWrapper fscdriver.SQLErrorWrapper) *TokenLockStore { return &TokenLockStore{ - ReadDB: readDB, - WriteDB: writeDB, - Table: tables, - Logger: logger, - ci: ci, + ReadDB: readDB, + WriteDB: writeDB, + Table: tables, + Logger: logger, + ci: ci, + errorWrapper: errorWrapper, } } -func NewTokenLockStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter) (*TokenLockStore, error) { +func NewTokenLockStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.CondInterpreter, errorWrapper fscdriver.SQLErrorWrapper) (*TokenLockStore, error) { return newTokenLockStore( readDB, writeDB, @@ -56,7 +60,9 @@ func NewTokenLockStore(readDB, writeDB *sql.DB, tables TableNames, ci common3.Co Tokens: tables.Tokens, Requests: tables.Requests, }, - ci), nil + ci, + errorWrapper, + ), nil } func (db *TokenLockStore) CreateSchema() error { @@ -67,12 +73,22 @@ func (db *TokenLockStore) CreateSchema() error { // selected for; this SQL-backed store does not apply per-wallet rate limiting and so // ignores it. A custom TokenLockStore may use walletID to throttle per wallet. func (db *TokenLockStore) Lock(ctx context.Context, tokenID *token.ID, consumerTxID transaction.ID, walletID string) error { + return db.LockAt(ctx, tokenID, consumerTxID, walletID, time.Now().UTC()) +} + +// 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). +func (db *TokenLockStore) LockAt(ctx context.Context, tokenID *token.ID, consumerTxID transaction.ID, _ string, createdAt time.Time) error { query, args := q.InsertInto(db.Table.TokenLocks). Fields("consumer_tx_id", "tx_id", "idx", "created_at"). - Row(consumerTxID, tokenID.TxId, tokenID.Index, time.Now().UTC()). + Row(consumerTxID, tokenID.TxId, tokenID.Index, createdAt). Format() logging.Debug(logger, query, tokenID, consumerTxID) _, err := db.WriteDB.ExecContext(ctx, query, args...) + if err != nil && errors.Is(db.errorWrapper.WrapError(err), fscdriver.UniqueKeyViolation) { + return errors.Wrapf(driver.ErrTokenAlreadyLocked, "token %s is already locked", tokenID) + } return err } @@ -95,7 +111,7 @@ func (db *TokenLockStore) GetSchema() string { tx_id TEXT NOT NULL, idx INT NOT NULL, consumer_tx_id TEXT NOT NULL, - created_at TIMESTAMP NOT NULL, + created_at TIMESTAMPTZ NOT NULL, PRIMARY KEY(tx_id, idx), FOREIGN KEY (tx_id, idx) REFERENCES %s ); @@ -117,3 +133,49 @@ 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. +// +// The argument order matches IsExpiredToken (tokenRequests first, tokenLocks second) +// so a swapped call is immediately visible and consistent across this file. +func IsStaleLock(tokenRequests, tokenLocks 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. created_at is declared TIMESTAMPTZ so the comparison with the +// database-side NOW() expression is always timezone-consistent on Postgres. +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(tokenRequests, tokenLocks, 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 +} diff --git a/token/services/storage/db/sql/postgres/tokenlock.go b/token/services/storage/db/sql/postgres/tokenlock.go index 6f5f583b58..4aa0713f28 100644 --- a/token/services/storage/db/sql/postgres/tokenlock.go +++ b/token/services/storage/db/sql/postgres/tokenlock.go @@ -15,6 +15,7 @@ import ( "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators" common2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/common" "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/common" + fscPostgres "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/postgres" "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" common5 "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/common" @@ -59,7 +60,7 @@ func (s *TokenLockStore) CreateSchema() error { // NewTokenLockStore returns a new TokenLockStore for the given RWDB and table names. func NewTokenLockStore(dbs *common2.RWDB, tableNames common5.TableNames) (*TokenLockStore, error) { ci := NewConditionInterpreter() - tldb, err := common5.NewTokenLockStore(dbs.ReadDB, dbs.WriteDB, tableNames, ci) + tldb, err := common5.NewTokenLockStore(dbs.ReadDB, dbs.WriteDB, tableNames, ci, &fscPostgres.ErrorMapper{}) if err != nil { return nil, err } @@ -85,40 +86,20 @@ 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. +// NOW() returns timestamptz; created_at is also TIMESTAMPTZ, so both sides of +// the age comparison are timezone-consistent. func (db *TokenLockStore) logStaleLocks(ctx context.Context, leaseExpiry time.Duration) error { if !db.Logger.IsEnabledFor(zapcore.InfoLevel) { return nil diff --git a/token/services/storage/db/sql/sqlite/tokenlock.go b/token/services/storage/db/sql/sqlite/tokenlock.go index 50b2fccccb..3f40aa1918 100644 --- a/token/services/storage/db/sql/sqlite/tokenlock.go +++ b/token/services/storage/db/sql/sqlite/tokenlock.go @@ -8,56 +8,18 @@ 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" + fscSqlite "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/storage/driver/sql/sqlite" "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 +29,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()) + tldb, err := common4.NewTokenLockStore(dbs.ReadDB, dbs.WriteDB, tableNames, NewConditionInterpreter(), &fscSqlite.ErrorMapper{}) if err != nil { return nil, err } - return &TokenLockStore{TokenLockStore: tldb, ci: NewConditionInterpreter()}, nil + return &TokenLockStore{TokenLockStore: tldb}, nil } diff --git a/token/services/storage/db/sql/sqlite/tokenlock_test.go b/token/services/storage/db/sql/sqlite/tokenlock_test.go index d263a436d3..2525a8c9c3 100644 --- a/token/services/storage/db/sql/sqlite/tokenlock_test.go +++ b/token/services/storage/db/sql/sqlite/tokenlock_test.go @@ -33,33 +33,26 @@ func mockTokenLockStore(db *sql.DB) *common3.TokenLockStore { return store.TokenLockStore } -func TestIsStale(t *testing.T) { +// TestCleanupSQLShape guards the shape of the cleanup statement rendered with the +// SQLite interpreter: the status branch must correlate on the consuming transaction +// and the deletion must not be scoped by a partial-key IN on tx_id. It asserts the +// parts that carry the semantics rather than the whole string - the string equality +// this replaced encoded a join on the wrong column as the expected output. The +// behaviour itself is covered by the shared suite in dbtest. See #2018. +func TestCleanupSQLShape(t *testing.T) { RegisterTestingT(t) + tokenLocks, requests := q.Table("TokenLocks"), q.Table("Requests") query, args := q.DeleteFrom("TokenLocks"). - Where(IsStale("TokenLocks", "Requests", 5*time.Second)). + Where(common3.IsStaleLock(requests, tokenLocks, 5*time.Second)). Format(NewConditionInterpreter()) - Expect(query).To(Equal("DELETE FROM TokenLocks WHERE tx_id IN (" + - "SELECT tl.tx_id " + - "FROM TokenLocks AS tl " + - "LEFT JOIN Requests AS tr " + - "ON tl.tx_id = tr.tx_id " + - "WHERE ((tr.status) IN (($1), ($2))) OR (tl.created_at < datetime('now', '-5 seconds'))" + - ")")) - Expect(args).To(ConsistOf(driver.Deleted, driver.Orphan)) -} - -func TestIsStaleOrphan(t *testing.T) { - RegisterTestingT(t) - - query, args := q.DeleteFrom("TokenLocks"). - Where(IsStale("TokenLocks", "Requests", 10*time.Second)). - Format(NewConditionInterpreter()) - - // Verify the query includes both Deleted (3) and Orphan (4) statuses using IN syntax - Expect(query).To(ContainSubstring("(tr.status) IN")) - Expect(query).To(ContainSubstring("datetime('now', '-10 seconds')")) + Expect(query).To(ContainSubstring("Requests.tx_id = TokenLocks.consumer_tx_id")) + Expect(query).To(ContainSubstring("EXISTS (SELECT 1 FROM Requests WHERE")) + Expect(query).To(ContainSubstring("(Requests.status) IN")) + Expect(query).To(ContainSubstring("TokenLocks.created_at < datetime('now', '-5 seconds')")) + Expect(query).ToNot(ContainSubstring("tx_id IN (")) + Expect(query).ToNot(ContainSubstring("TokenLocks.tx_id = Requests.tx_id")) Expect(args).To(ConsistOf(driver.Deleted, driver.Orphan)) }