Implements token lock cleanup for orphan transactions

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),
 	)
 }

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()
 

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())
+}

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) {