From e296b4c35b968482836ae1ab2b057bc7070de662 Mon Sep 17 00:00:00 2001 From: AkramBitar Date: Thu, 25 Jun 2026 07:42:11 +0000 Subject: [PATCH] Implements token lock cleanup for orphan transactions, treating them the same as deleted transactions for immediate lock release. Signed-off-by: AkramBitar --- .../storage/db/sql/common/tokenlock.go | 2 +- .../storage/db/sql/postgres/tokenlock.go | 4 +- .../storage/db/sql/postgres/tokenlock_test.go | 48 +++++++++++++++++-- .../storage/db/sql/sqlite/tokenlock_test.go | 17 ++++++- 4 files changed, 61 insertions(+), 10 deletions(-) diff --git a/token/services/storage/db/sql/common/tokenlock.go b/token/services/storage/db/sql/common/tokenlock.go index 220d45da27..e8874e28a1 100644 --- a/token/services/storage/db/sql/common/tokenlock.go +++ b/token/services/storage/db/sql/common/tokenlock.go @@ -107,7 +107,7 @@ func (db *TokenLockStore) Close() error { func IsExpiredToken(tokenRequests, tokenLocks common3.Table, leaseExpiry time.Duration) cond.Condition { return cond.Or( - cond.FieldIn(tokenRequests.Field("status"), driver.Deleted), + cond.FieldIn(tokenRequests.Field("status"), driver.Deleted, driver.Orphan), cond.OlderThan(tokenLocks.Field("created_at"), leaseExpiry), ) } diff --git a/token/services/storage/db/sql/postgres/tokenlock.go b/token/services/storage/db/sql/postgres/tokenlock.go index b7a863cf0a..cedac460c2 100644 --- a/token/services/storage/db/sql/postgres/tokenlock.go +++ b/token/services/storage/db/sql/postgres/tokenlock.go @@ -76,8 +76,8 @@ func (db *TokenLockStore) Cleanup(ctx context.Context, leaseExpiry time.Duration WriteConditionSerializable(cond.OlderThan(tokenLocks.Field("created_at"), leaseExpiry), db.ci). WriteString(" OR "). WriteString( - fmt.Sprintf("EXISTS (SELECT 1 FROM %s WHERE %s.tx_id = %s.consumer_tx_id AND %s.status IN (%d))", - db.Table.Requests, db.Table.Requests, db.Table.TokenLocks, db.Table.Requests, driver.Deleted, + fmt.Sprintf("EXISTS (SELECT 1 FROM %s WHERE %s.tx_id = %s.consumer_tx_id AND %s.status IN (%d, %d))", + db.Table.Requests, db.Table.Requests, db.Table.TokenLocks, db.Table.Requests, driver.Deleted, driver.Orphan, )). // TODO: Implement EXIST condition Build() diff --git a/token/services/storage/db/sql/postgres/tokenlock_test.go b/token/services/storage/db/sql/postgres/tokenlock_test.go index 49e1e29ead..5d70f5a4f3 100644 --- a/token/services/storage/db/sql/postgres/tokenlock_test.go +++ b/token/services/storage/db/sql/postgres/tokenlock_test.go @@ -65,16 +65,16 @@ func TestCleanup(t *testing.T) { consumerTxID, tokenID.TxId, tokenID.Index, *status, createdAt, timeNow, } mockDB. - ExpectQuery("SELECT TOKEN_LOCKS.consumer_tx_id, TOKEN_LOCKS.tx_id, TOKEN_LOCKS.idx, REQUESTS.status, TOKEN_LOCKS.created_at, NOW\\(\\) AS now " + - "FROM TOKEN_LOCKS LEFT JOIN REQUESTS ON TOKEN_LOCKS.consumer_tx_id = REQUESTS.tx_id " + - "WHERE \\(\\(\\(REQUESTS.status = \\$1\\)\\)\\) OR \\(TOKEN_LOCKS.created_at < NOW\\(\\) - INTERVAL '1 seconds'\\)"). - WithArgs(input). + ExpectQuery("SELECT TOKEN_LOCKS.consumer_tx_id, TOKEN_LOCKS.tx_id, TOKEN_LOCKS.idx, REQUESTS.status, TOKEN_LOCKS.created_at, NOW\\(\\) AS now "+ + "FROM TOKEN_LOCKS LEFT JOIN REQUESTS ON TOKEN_LOCKS.consumer_tx_id = REQUESTS.tx_id "+ + "WHERE \\(\\(\\(REQUESTS.status = \\$1\\)\\) OR \\(\\(REQUESTS.status = \\$2\\)\\)\\) OR \\(TOKEN_LOCKS.created_at < NOW\\(\\) - INTERVAL '1 seconds'\\)"). + WithArgs(driver.Deleted, driver.Orphan). WillReturnRows(mockDB.NewRows([]string{"consumer_tx_id", "tx_id", "idx", "status", "created_at", "now"}).AddRow(output...)) mockDB.ExpectExec("DELETE FROM TOKEN_LOCKS WHERE " + "TOKEN_LOCKS.created_at < NOW\\(\\) - INTERVAL '1 seconds'" + " OR " + - "EXISTS \\(SELECT 1 FROM REQUESTS WHERE REQUESTS.tx_id = TOKEN_LOCKS.consumer_tx_id AND REQUESTS.status IN \\(3\\)"). + "EXISTS \\(SELECT 1 FROM REQUESTS WHERE REQUESTS.tx_id = TOKEN_LOCKS.consumer_tx_id AND REQUESTS.status IN \\(3, 4\\)"). WillReturnResult(sqlmock.NewResult(0, 1)) err = mockTokenLockStorePostgress(db).Cleanup(t.Context(), time.Second) @@ -90,3 +90,41 @@ func TestLock(t *testing.T) { func TestUnlockByTxID(t *testing.T) { common3.TestUnlockByTxID(t, mockTokenLockStore) } + +func TestCleanupOrphan(t *testing.T) { + RegisterTestingT(t) + db, mockDB, err := sqlmock.New() + Expect(err).ToNot(HaveOccurred()) + + input := driver.Orphan + + consumerTxID := "orphan-tx-1234" + tokenID := token.ID{TxId: "5678", Index: 5} + status := &input + createdAt := time.Date(2025, time.June, 8, 10, 0, 0, 0, time.UTC) + + var timeNow time.Time + output := []driver2.Value{ + consumerTxID, tokenID.TxId, tokenID.Index, *status, createdAt, timeNow, + } + + // Expect the SELECT query for logging - now includes both Deleted and Orphan + mockDB. + ExpectQuery("SELECT TOKEN_LOCKS.consumer_tx_id, TOKEN_LOCKS.tx_id, TOKEN_LOCKS.idx, REQUESTS.status, TOKEN_LOCKS.created_at, NOW\\(\\) AS now "+ + "FROM TOKEN_LOCKS LEFT JOIN REQUESTS ON TOKEN_LOCKS.consumer_tx_id = REQUESTS.tx_id "+ + "WHERE \\(\\(\\(REQUESTS.status = \\$1\\)\\) OR \\(\\(REQUESTS.status = \\$2\\)\\)\\) OR \\(TOKEN_LOCKS.created_at < NOW\\(\\) - INTERVAL '1 seconds'\\)"). + WithArgs(driver.Deleted, driver.Orphan). + WillReturnRows(mockDB.NewRows([]string{"consumer_tx_id", "tx_id", "idx", "status", "created_at", "now"}).AddRow(output...)) + + // Expect the DELETE query with both Deleted (3) and Orphan (4) statuses + mockDB.ExpectExec("DELETE FROM TOKEN_LOCKS WHERE " + + "TOKEN_LOCKS.created_at < NOW\\(\\) - INTERVAL '1 seconds'" + + " OR " + + "EXISTS \\(SELECT 1 FROM REQUESTS WHERE REQUESTS.tx_id = TOKEN_LOCKS.consumer_tx_id AND REQUESTS.status IN \\(3, 4\\)"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + err = mockTokenLockStorePostgress(db).Cleanup(t.Context(), time.Second) + + Expect(mockDB.ExpectationsWereMet()).To(Succeed()) + Expect(err).ToNot(HaveOccurred()) +} diff --git a/token/services/storage/db/sql/sqlite/tokenlock_test.go b/token/services/storage/db/sql/sqlite/tokenlock_test.go index 4906dec10f..93d8364feb 100644 --- a/token/services/storage/db/sql/sqlite/tokenlock_test.go +++ b/token/services/storage/db/sql/sqlite/tokenlock_test.go @@ -45,9 +45,22 @@ func TestIsStale(t *testing.T) { "FROM TokenLocks AS tl " + "LEFT JOIN Requests AS tr " + "ON tl.tx_id = tr.tx_id " + - "WHERE (tr.status = $1) OR (tl.created_at < datetime('now', '-5 seconds'))" + + "WHERE ((tr.status) IN (($1), ($2))) OR (tl.created_at < datetime('now', '-5 seconds'))" + ")")) - Expect(args).To(ConsistOf(driver.Deleted)) + 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(args).To(ConsistOf(driver.Deleted, driver.Orphan)) } func TestLock(t *testing.T) {