Only cleanup orphan upstream authorization sessions

This includes sessions that were never completed, and sessions where
user_session was cleaned up. This is to avoid breaking features like
OIDC Backchannel Logout after 30 days.

crates/storage-pg/.sqlx/query-6db23fc9c39c2c7d9224d4e1233205f636568c990ccb05cf9208750ad1330b9b.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n                WITH to_delete AS (\n                    SELECT upstream_oauth_authorization_session_id\n                    FROM upstream_oauth_authorization_sessions\n                    WHERE ($1::uuid IS NULL OR upstream_oauth_authorization_session_id > $1)\n                      AND upstream_oauth_authorization_session_id <= $2\n                    ORDER BY upstream_oauth_authorization_session_id\n                    LIMIT $3\n                )\n                DELETE FROM upstream_oauth_authorization_sessions\n                USING to_delete\n                WHERE upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id = to_delete.upstream_oauth_authorization_session_id\n                RETURNING upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id\n            ",
+  "query": "\n                WITH to_delete AS (\n                    SELECT upstream_oauth_authorization_session_id\n                    FROM upstream_oauth_authorization_sessions\n                    WHERE ($1::uuid IS NULL OR upstream_oauth_authorization_session_id > $1)\n                      AND upstream_oauth_authorization_session_id <= $2\n                      AND user_session_id IS NULL\n                    ORDER BY upstream_oauth_authorization_session_id\n                    LIMIT $3\n                )\n                DELETE FROM upstream_oauth_authorization_sessions\n                USING to_delete\n                WHERE upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id = to_delete.upstream_oauth_authorization_session_id\n                RETURNING upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id\n            ",
   "describe": {
     "columns": [
       {
@@ -20,5 +20,5 @@
       false
     ]
   },
-  "hash": "da6baa340eedfce8e965c9f3baa90f21f2331c3881c082f0157752d241403b35"
+  "hash": "6db23fc9c39c2c7d9224d4e1233205f636568c990ccb05cf9208750ad1330b9b"
 }

crates/storage-pg/migrations/20260121112201_upstream_oauth_sessions_orphan_index.sql

@@ -0,0 +1,10 @@
+-- no-transaction
+-- Copyright 2026 Element Creations Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+-- Please see LICENSE files in the repository root for full details.
+
+-- Add partial index for cleanup of orphaned upstream OAuth sessions
+CREATE INDEX CONCURRENTLY IF NOT EXISTS upstream_oauth_authorization_sessions_orphaned
+    ON upstream_oauth_authorization_sessions (upstream_oauth_authorization_session_id)
+    WHERE user_session_id IS NULL;

crates/storage-pg/src/upstream_oauth2/session.rs

@@ -580,7 +580,7 @@ impl UpstreamOAuthSessionRepository for PgUpstreamOAuthSessionRepository<'_> {
         ),
         err,
     )]
-    async fn cleanup(
+    async fn cleanup_orphaned(
         &mut self,
         since: Option<Ulid>,
         until: Ulid,
@@ -596,6 +596,7 @@ impl UpstreamOAuthSessionRepository for PgUpstreamOAuthSessionRepository<'_> {
                     FROM upstream_oauth_authorization_sessions
                     WHERE ($1::uuid IS NULL OR upstream_oauth_authorization_session_id > $1)
                       AND upstream_oauth_authorization_session_id <= $2
+                      AND user_session_id IS NULL
                     ORDER BY upstream_oauth_authorization_session_id
                     LIMIT $3
                 )

crates/storage/src/upstream_oauth2/session.rs

@@ -211,9 +211,11 @@ pub trait UpstreamOAuthSessionRepository: Send + Sync {
     async fn count(&mut self, filter: UpstreamOAuthSessionFilter<'_>)
     -> Result<usize, Self::Error>;
 
-    /// Cleanup old authorization sessions
+    /// Cleanup old authorization sessions that are not linked to a user session
     ///
     /// This will delete sessions with IDs up to and including `until`.
+    /// Authorization sessions with a user session linked must be kept around to
+    /// avoid breaking features like OIDC Backchannel Logout.
     ///
     /// Returns the number of sessions deleted and the cursor for the next batch
     ///
@@ -227,7 +229,7 @@ pub trait UpstreamOAuthSessionRepository: Send + Sync {
     /// # Errors
     ///
     /// Returns [`Self::Error`] if the underlying repository fails
-    async fn cleanup(
+    async fn cleanup_orphaned(
         &mut self,
         since: Option<Ulid>,
         until: Ulid,
@@ -277,7 +279,7 @@ repository_impl!(UpstreamOAuthSessionRepository:
 
     async fn count(&mut self, filter: UpstreamOAuthSessionFilter<'_>) -> Result<usize, Self::Error>;
 
-    async fn cleanup(
+    async fn cleanup_orphaned(
         &mut self,
         since: Option<Ulid>,
         until: Ulid,

crates/tasks/src/database.rs

@@ -345,7 +345,7 @@ impl RunnableJob for CleanupUpstreamOAuthSessionsJob {
             let mut repo = state.repository().await.map_err(JobError::retry)?;
             let (count, cursor) = repo
                 .upstream_oauth_session()
-                .cleanup(since, until, BATCH_SIZE)
+                .cleanup_orphaned(since, until, BATCH_SIZE)
                 .await
                 .map_err(JobError::retry)?;
             repo.save().await.map_err(JobError::retry)?;