feat(oidc): use worker queue for OIDC Back-Channel Logout notifications (#11441)

# Which Problems Are Solved

The current back-channel logout notification still used a single
projection handler for handling terminating sessions and notifying the
necessary clients. This can lead to back pressure and delayed
notifications.

# How the Problems Are Solved

The handler now only creates a job in the (river) worker queue to handle
the notification.
The worker will then search for all necessary clients to be informed and
create a job for each.
These jobs will then be picked up by the (same) worker(s) again, which
will create, sign and send the logout token to the client. If one
notification fails, this one job will be retired, but others can still
be processed in parallel and ore not affected.
Each successful job will still create an event on the session stating
the successful notification of the client.
The existing `OIDC.DefaultBackChannelLogoutLifetime` configuration has
been deprecated in favor of making it part of the new
`OIDC.BackChannelLogout` config.
 
# Additional Changes

None

# Additional Context

- feature check is still executed and will be removed seperately for
easier review: https://github.com/zitadel/zitadel/issues/11277
- closes https://github.com/zitadel/zitadel/issues/9279
- requires backport to v4.x

cmd/defaults.yaml

@@ -625,8 +625,22 @@ OIDC:
   DefaultLogoutURLV2: "/ui/v2/login/logout?post_logout_redirect=" # ZITADEL_OIDC_DEFAULTLOGOUTURLV2
   # Internal cache age for public keys to speed up validations (e.g. id_token_hints) on the authorization endpoint.
   PublicKeyCacheMaxAge: 24h # ZITADEL_OIDC_PUBLICKEYCACHEMAXAGE
-  # Lifetime of the token used to notify clients through OIDC back-channel logout when the feature is enabled and registered on the client.
+  # Lifetime of the token used to notify clients through OIDC back-channel logout.
+  # Deprecated: use BackChannelLogout.TokenLifetime instead
   DefaultBackChannelLogoutLifetime: 15m # ZITADEL_OIDC_DEFAULTBACKCHANNELLOGOUTLIFETIME
+  BackChannelLogout:
+    # The amount of workers processing the back-channel logout notifications.
+    # If set to 0, no notifications will be handled. This can be useful when running in
+    # multi binary / pod setup and allowing only certain executables to process the events.
+    Workers: 1 # ZITADEL_OIDC_BACKCHANNELLOGOUT_WORKERS
+    # The maximum duration a job can do it's work before it is considered as failed.
+    TransactionDuration: 10s # ZITADEL_OIDC_BACKCHANNELLOGOUT_TRANSACTIONDURATION
+    # Automatically cancel the notification after the amount of failed attempts
+    MaxAttempts: 3 # ZITADEL_OIDC_BACKCHANNELLOGOUT_MAXATTEMPTS
+    # Automatically cancel the notification if it cannot be handled within a specific time
+    MaxTtl: 5m  # ZITADEL_OIDC_BACKCHANNELLOGOUT_MAXTTL
+    # Lifetime of the token used to notify clients through OIDC back-channel logout.
+    TokenLifetime: 15m # ZITADEL_OIDC_BACKCHANNELLOGOUT_TOKENLIFETIME
 
 SAML:
   DefaultLoginURLV2: "/ui/v2/login/login?samlRequest=" # ZITADEL_SAML_DEFAULTLOGINURLV2

cmd/mirror/projections.go

@@ -216,6 +216,7 @@ func projections(
 		config.Projections.Customizations["backchannel"],
 		config.Projections.Customizations["telemetry"],
 		config.Notifications,
+		config.OIDC.BackChannelLogoutConfig(),
 		*config.Telemetry,
 		config.ExternalDomain,
 		config.ExternalPort,
@@ -228,8 +229,6 @@ func projections(
 		keys.User,
 		keys.SMTP,
 		keys.SMS,
-		keys.OIDC,
-		config.OIDC.DefaultBackChannelLogoutLifetime,
 		nil,
 	)
 

cmd/setup/setup.go

@@ -589,6 +589,7 @@ func startCommandsQueries(
 		config.Projections.Customizations["backchannel"],
 		config.Projections.Customizations["telemetry"],
 		config.Notifications,
+		config.OIDC.BackChannelLogoutConfig(),
 		*config.Telemetry,
 		config.ExternalDomain,
 		config.ExternalPort,
@@ -601,8 +602,6 @@ func startCommandsQueries(
 		keys.User,
 		keys.SMTP,
 		keys.SMS,
-		keys.OIDC,
-		config.OIDC.DefaultBackChannelLogoutLifetime,
 		q,
 	)
 

cmd/start/start.go

@@ -324,6 +324,7 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server
 		config.Projections.Customizations["backchannel"],
 		config.Projections.Customizations["telemetry"],
 		config.Notifications,
+		config.OIDC.BackChannelLogoutConfig(),
 		*config.Telemetry,
 		config.ExternalDomain,
 		config.ExternalPort,
@@ -336,8 +337,6 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server
 		keys.User,
 		keys.SMTP,
 		keys.SMS,
-		keys.OIDC,
-		config.OIDC.DefaultBackChannelLogoutLifetime,
 		q,
 	)
 	notification.Start(ctx)

internal/api/oidc/op.go

@@ -21,6 +21,7 @@ import (
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/domain/federatedlogout"
 	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/notification/handlers"
 	"github.com/zitadel/zitadel/internal/query"
 	"github.com/zitadel/zitadel/internal/zerrors"
 )
@@ -42,6 +43,16 @@ type Config struct {
 	DefaultLogoutURLV2                string
 	PublicKeyCacheMaxAge              time.Duration
 	DefaultBackChannelLogoutLifetime  time.Duration
+	BackChannelLogout                 handlers.BackChannelLogoutWorkerConfig
+}
+
+// BackChannelLogoutConfig returns the BackChannelLogoutWorkerConfig and takes the deprecated TokenLifetime into account.
+func (c *Config) BackChannelLogoutConfig() *handlers.BackChannelLogoutWorkerConfig {
+	if c.DefaultBackChannelLogoutLifetime == 0 {
+		return &c.BackChannelLogout
+	}
+	c.BackChannelLogout.TokenLifetime = c.DefaultBackChannelLogoutLifetime
+	return &c.BackChannelLogout
 }
 
 type EndpointConfig struct {

internal/notification/backchannel/back_channel_logout.go

@@ -0,0 +1,23 @@
+package backchannel
+
+import "github.com/zitadel/zitadel/internal/eventstore"
+
+const (
+	QueueName = "back_channel_logout"
+)
+
+type LogoutRequest struct {
+	Aggregate            *eventstore.Aggregate
+	SessionID            string
+	TriggeredAtOrigin    string
+	TriggeringEventType  eventstore.EventType
+	TokenID              string
+	UserID               string
+	OIDCSessionID        string
+	ClientID             string
+	BackChannelLogoutURI string
+}
+
+func (l *LogoutRequest) Kind() string {
+	return "back_channel_logout_request"
+}

internal/notification/handlers/back_channel_logout.go

@@ -2,25 +2,15 @@ package handlers
 
 import (
 	"context"
-	"errors"
 	"slices"
-	"sync"
-	"time"
-
-	"github.com/zitadel/oidc/v3/pkg/crypto"
-	"github.com/zitadel/oidc/v3/pkg/oidc"
 
 	"github.com/zitadel/zitadel/internal/api/authz"
-	http_utils "github.com/zitadel/zitadel/internal/api/http"
-	"github.com/zitadel/zitadel/internal/api/oidc/sign"
-	"github.com/zitadel/zitadel/internal/command"
-	zcrypto "github.com/zitadel/zitadel/internal/crypto"
+	http_util "github.com/zitadel/zitadel/internal/api/http"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
-	"github.com/zitadel/zitadel/internal/id"
-	"github.com/zitadel/zitadel/internal/notification/channels/set"
+	"github.com/zitadel/zitadel/internal/notification/backchannel"
 	_ "github.com/zitadel/zitadel/internal/notification/statik"
-	"github.com/zitadel/zitadel/internal/notification/types"
+	"github.com/zitadel/zitadel/internal/queue"
 	"github.com/zitadel/zitadel/internal/repository/session"
 	"github.com/zitadel/zitadel/internal/repository/sessionlogout"
 	"github.com/zitadel/zitadel/internal/repository/user"
@@ -32,33 +22,22 @@ const (
 )
 
 type backChannelLogoutNotifier struct {
-	commands         *command.Commands
-	queries          *NotificationQueries
-	eventstore       *eventstore.Eventstore
-	keyEncryptionAlg zcrypto.EncryptionAlgorithm
-	channels         types.ChannelChains
-	idGenerator      id.Generator
-	tokenLifetime    time.Duration
+	queries     *NotificationQueries
+	queue       Queue
+	maxAttempts uint8
 }
 
 func NewBackChannelLogoutNotifier(
 	ctx context.Context,
 	config handler.Config,
-	commands *command.Commands,
 	queries *NotificationQueries,
-	es *eventstore.Eventstore,
-	keyEncryptionAlg zcrypto.EncryptionAlgorithm,
-	channels types.ChannelChains,
-	tokenLifetime time.Duration,
+	queue Queue,
+	maxAttempts uint8,
 ) *handler.Handler {
 	return handler.NewHandler(ctx, &config, &backChannelLogoutNotifier{
-		commands:         commands,
-		queries:          queries,
-		eventstore:       es,
-		keyEncryptionAlg: keyEncryptionAlg,
-		channels:         channels,
-		tokenLifetime:    tokenLifetime,
-		idGenerator:      id.SonyFlakeGenerator(),
+		queries:     queries,
+		queue:       queue,
+		maxAttempts: maxAttempts,
 	})
 
 }
@@ -106,7 +85,20 @@ func (u *backChannelLogoutNotifier) reduceUserSignedOut(event eventstore.Event)
 		if e.SessionID == "" {
 			return nil
 		}
-		return u.terminateSession(ctx, e.SessionID, e)
+		ctx, err = u.queries.Origin(ctx, e)
+		if err != nil {
+			return err
+		}
+		return u.queue.Insert(ctx,
+			&backchannel.LogoutRequest{
+				Aggregate:           e.Aggregate(),
+				SessionID:           e.SessionID,
+				TriggeredAtOrigin:   http_util.DomainContext(ctx).Origin(),
+				TriggeringEventType: event.Type(),
+			},
+			queue.WithQueueName(backchannel.QueueName),
+			queue.WithMaxAttempts(u.maxAttempts),
+		)
 	}), nil
 }
 
@@ -124,7 +116,20 @@ func (u *backChannelLogoutNotifier) reduceSessionTerminated(event eventstore.Eve
 		if !authz.GetFeatures(ctx).EnableBackChannelLogout {
 			return nil
 		}
-		return u.terminateSession(ctx, e.Aggregate().ID, e)
+		ctx, err = u.queries.Origin(ctx, e)
+		if err != nil {
+			return err
+		}
+		return u.queue.Insert(ctx,
+			&backchannel.LogoutRequest{
+				Aggregate:           e.Aggregate(),
+				SessionID:           e.Aggregate().ID,
+				TriggeredAtOrigin:   http_util.DomainContext(ctx).Origin(),
+				TriggeringEventType: event.Type(),
+			},
+			queue.WithQueueName(backchannel.QueueName),
+			queue.WithMaxAttempts(u.maxAttempts),
+		)
 	}), nil
 }
 
@@ -135,74 +140,6 @@ type backChannelLogoutSession struct {
 	sessions []backChannelLogoutOIDCSessions
 }
 
-func (u *backChannelLogoutNotifier) terminateSession(ctx context.Context, id string, e eventstore.Event) error {
-	sessions := &backChannelLogoutSession{sessionID: id}
-	err := u.eventstore.FilterToQueryReducer(ctx, sessions)
-	if err != nil {
-		return err
-	}
-
-	ctx, err = u.queries.Origin(ctx, e)
-	if err != nil {
-		return err
-	}
-
-	getSigner := sign.GetSignerOnce(u.queries.GetActiveSigningWebKey)
-
-	var wg sync.WaitGroup
-	wg.Add(len(sessions.sessions))
-	errs := make([]error, 0, len(sessions.sessions))
-	for _, oidcSession := range sessions.sessions {
-		go func(oidcSession *backChannelLogoutOIDCSessions) {
-			defer wg.Done()
-			err := u.sendLogoutToken(ctx, oidcSession, e, getSigner)
-			if err != nil {
-				errs = append(errs, err)
-				return
-			}
-			err = u.commands.BackChannelLogoutSent(ctx, oidcSession.SessionID, oidcSession.OIDCSessionID, e.Aggregate().InstanceID)
-			if err != nil {
-				errs = append(errs, err)
-			}
-		}(&oidcSession)
-	}
-	wg.Wait()
-	return errors.Join(errs...)
-}
-
-func (u *backChannelLogoutNotifier) sendLogoutToken(ctx context.Context, oidcSession *backChannelLogoutOIDCSessions, e eventstore.Event, getSigner sign.SignerFunc) error {
-	token, err := u.logoutToken(ctx, oidcSession, getSigner)
-	if err != nil {
-		return err
-	}
-	err = types.SendSecurityTokenEvent(ctx, set.Config{CallURL: oidcSession.BackChannelLogoutURI}, u.channels, &LogoutTokenMessage{LogoutToken: token}, e.Type()).WithoutTemplate()
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-func (u *backChannelLogoutNotifier) logoutToken(ctx context.Context, oidcSession *backChannelLogoutOIDCSessions, getSigner sign.SignerFunc) (string, error) {
-	jwtID, err := u.idGenerator.Next()
-	if err != nil {
-		return "", err
-	}
-	token := oidc.NewLogoutTokenClaims(
-		http_utils.DomainContext(ctx).Origin(),
-		oidcSession.UserID,
-		oidc.Audience{oidcSession.ClientID},
-		time.Now().Add(u.tokenLifetime),
-		jwtID,
-		oidcSession.SessionID,
-		time.Second,
-	)
-	signer, _, err := getSigner(ctx)
-	if err != nil {
-		return "", err
-	}
-	return crypto.Sign(token, signer)
-}
-
 type LogoutTokenMessage struct {
 	LogoutToken string `schema:"logout_token"`
 }

internal/notification/handlers/back_channel_logout_worker.go

@@ -0,0 +1,160 @@
+package handlers
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/riverqueue/river"
+	"github.com/zitadel/oidc/v3/pkg/crypto"
+	"github.com/zitadel/oidc/v3/pkg/oidc"
+
+	"github.com/zitadel/zitadel/internal/api/oidc/sign"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/id"
+	"github.com/zitadel/zitadel/internal/notification/backchannel"
+	"github.com/zitadel/zitadel/internal/notification/channels"
+	"github.com/zitadel/zitadel/internal/notification/channels/set"
+	"github.com/zitadel/zitadel/internal/notification/types"
+	"github.com/zitadel/zitadel/internal/queue"
+)
+
+type BackChannelLogoutWorker struct {
+	river.WorkerDefaults[*backchannel.LogoutRequest]
+
+	commands    Commands
+	queries     *NotificationQueries
+	eventstore  *eventstore.Eventstore
+	queue       Queue
+	channels    types.ChannelChains
+	config      *BackChannelLogoutWorkerConfig
+	now         nowFunc
+	idGenerator id.Generator
+}
+
+// Timeout implements the Timeout-function of [river.Worker].
+// Maximum time a job can run before the context gets canceled.
+func (w *BackChannelLogoutWorker) Timeout(*river.Job[*backchannel.LogoutRequest]) time.Duration {
+	return w.config.TransactionDuration
+}
+
+// Work implements [river.Worker].
+func (w *BackChannelLogoutWorker) Work(ctx context.Context, job *river.Job[*backchannel.LogoutRequest]) error {
+	ctx = ContextWithNotifier(ctx, job.Args.Aggregate)
+
+	ctx, err := enrichCtx(ctx, job.Args.TriggeredAtOrigin)
+	if err != nil {
+		return channels.NewCancelError(err)
+	}
+
+	// if the notification is too old, we can directly cancel
+	if job.CreatedAt.Add(w.config.MaxTtl).Before(w.now()) {
+		return river.JobCancel(errors.New("back channel logout notification is too old"))
+	}
+
+	if job.Args.OIDCSessionID == "" {
+		return w.createNotificationJobs(ctx, job.Args)
+	}
+	return w.sendLogoutRequest(ctx, job.Args)
+}
+
+func (w *BackChannelLogoutWorker) createNotificationJobs(ctx context.Context, request *backchannel.LogoutRequest) error {
+	sessions := &backChannelLogoutSession{sessionID: request.SessionID}
+	err := w.eventstore.FilterToQueryReducer(ctx, sessions)
+	if err != nil {
+		return err
+	}
+
+	for _, oidcSession := range sessions.sessions {
+		tokenID, err := w.idGenerator.Next()
+		if err != nil {
+			return err
+		}
+		logoutRequest := &backchannel.LogoutRequest{
+			Aggregate:            request.Aggregate,
+			SessionID:            request.SessionID,
+			TriggeredAtOrigin:    request.TriggeredAtOrigin,
+			TriggeringEventType:  request.TriggeringEventType,
+			TokenID:              tokenID,
+			UserID:               oidcSession.UserID,
+			OIDCSessionID:        oidcSession.OIDCSessionID,
+			ClientID:             oidcSession.ClientID,
+			BackChannelLogoutURI: oidcSession.BackChannelLogoutURI,
+		}
+		err = w.queue.Insert(ctx, logoutRequest,
+			queue.WithQueueName(backchannel.QueueName),
+			queue.WithMaxAttempts(w.config.MaxAttempts))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *BackChannelLogoutWorker) sendLogoutRequest(ctx context.Context, request *backchannel.LogoutRequest) error {
+	getSigner := sign.GetSignerOnce(w.queries.GetActiveSigningWebKey)
+	token, err := w.logoutToken(ctx, request, getSigner)
+	if err != nil {
+		return err
+	}
+	if err = types.SendSecurityTokenEvent(ctx, set.Config{CallURL: request.BackChannelLogoutURI}, w.channels, &LogoutTokenMessage{LogoutToken: token}, request.TriggeringEventType).WithoutTemplate(); err != nil {
+		return err
+	}
+	return w.commands.BackChannelLogoutSent(ctx, request.SessionID, request.OIDCSessionID, request.Aggregate.InstanceID)
+}
+
+func (w *BackChannelLogoutWorker) logoutToken(ctx context.Context, request *backchannel.LogoutRequest, getSigner sign.SignerFunc) (string, error) {
+	token := oidc.NewLogoutTokenClaims(
+		request.TriggeredAtOrigin,
+		request.UserID,
+		oidc.Audience{request.ClientID},
+		w.now().Add(w.config.TokenLifetime),
+		request.TokenID,
+		request.SessionID,
+		time.Second,
+	)
+	signer, _, err := getSigner(ctx)
+	if err != nil {
+		return "", err
+	}
+	return crypto.Sign(token, signer)
+}
+
+func NewBackChannelLogoutWorker(
+	commands Commands,
+	queries *NotificationQueries,
+	eventstore *eventstore.Eventstore,
+	queue Queue,
+	channels types.ChannelChains,
+	config *BackChannelLogoutWorkerConfig,
+
+	idGenerator id.Generator,
+) *BackChannelLogoutWorker {
+	return &BackChannelLogoutWorker{
+		commands:    commands,
+		queries:     queries,
+		eventstore:  eventstore,
+		queue:       queue,
+		channels:    channels,
+		config:      config,
+		now:         time.Now,
+		idGenerator: idGenerator,
+	}
+}
+
+var _ river.Worker[*backchannel.LogoutRequest] = (*BackChannelLogoutWorker)(nil)
+
+func (w *BackChannelLogoutWorker) Register(workers *river.Workers, queues map[string]river.QueueConfig) {
+	river.AddWorker(workers, w)
+	queues[backchannel.QueueName] = river.QueueConfig{
+		MaxWorkers: int(w.config.Workers),
+	}
+}
+
+type BackChannelLogoutWorkerConfig struct {
+	Workers             uint8
+	TransactionDuration time.Duration
+	MaxAttempts         uint8
+	MaxTtl              time.Duration
+	TokenLifetime       time.Duration
+}

internal/notification/handlers/back_channel_logout_worker_test.go

@@ -0,0 +1,508 @@
+package handlers
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"database/sql"
+	"encoding/json"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/go-jose/go-jose/v4"
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zitadel/oidc/v3/pkg/oidc"
+	"go.uber.org/mock/gomock"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/eventstore/repository"
+	es_repo_mock "github.com/zitadel/zitadel/internal/eventstore/repository/mock"
+	"github.com/zitadel/zitadel/internal/id"
+	id_mock "github.com/zitadel/zitadel/internal/id/mock"
+	"github.com/zitadel/zitadel/internal/notification/backchannel"
+	"github.com/zitadel/zitadel/internal/notification/channels"
+	channel_mock "github.com/zitadel/zitadel/internal/notification/channels/mock"
+	"github.com/zitadel/zitadel/internal/notification/handlers/mock"
+	"github.com/zitadel/zitadel/internal/notification/messages"
+	"github.com/zitadel/zitadel/internal/notification/senders"
+	"github.com/zitadel/zitadel/internal/queue"
+	"github.com/zitadel/zitadel/internal/repository/sessionlogout"
+)
+
+func Test_backChannelLogoutWorker_reduceNotificationRequested(t *testing.T) {
+	testNow := time.Now()
+	sessionLogoutAgg := &sessionlogout.NewAggregate(sessionID, instanceID).Aggregate
+	type fields struct {
+		es          func(*testing.T) *eventstore.Eventstore
+		queue       func(*gomock.Controller) Queue
+		commands    func(*gomock.Controller) Commands
+		queries     func(*gomock.Controller) Queries
+		channel     func(*gomock.Controller) channels.NotificationChannel
+		idGenerator id.Generator
+	}
+	type args struct {
+		job *river.Job[*backchannel.LogoutRequest]
+	}
+	type want struct {
+		err error
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   want
+	}{
+		{
+			name: "too old",
+			fields: fields{
+				es: expectEventstore(),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					return c
+				},
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: time.Now().Add(-1 * time.Hour),
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate: sessionLogoutAgg,
+					},
+				},
+			},
+			want: want{
+				err: new(river.JobCancelError),
+			},
+		},
+		{
+			name: "no oidc sessions with back channel logout",
+			fields: fields{
+				es: expectEventstore(
+					expectFilter(),
+				),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					return c
+				},
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: testNow,
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate: sessionLogoutAgg,
+						SessionID: sessionID,
+					},
+				},
+			},
+			want: want{
+				err: nil,
+			},
+		},
+		{
+			name: "create jobs for oidc sessions with back channel logout",
+			fields: fields{
+				es: expectEventstore(
+					expectFilter(
+						eventFromEventPusher(
+							sessionlogout.NewBackChannelLogoutRegisteredEvent(
+								context.Background(),
+								sessionLogoutAgg,
+								"oidc-session-id1",
+								"user-id",
+								"client-id1",
+								"back-channel-logout-uri1",
+							),
+						),
+						eventFromEventPusher(
+							sessionlogout.NewBackChannelLogoutRegisteredEvent(
+								context.Background(),
+								sessionLogoutAgg,
+								"oidc-session-id2",
+								"user-id",
+								"client-id2",
+								"back-channel-logout-uri2",
+							),
+						),
+					),
+				),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					q.EXPECT().Insert(gomock.Any(),
+						&backchannel.LogoutRequest{
+							Aggregate:            sessionLogoutAgg,
+							SessionID:            sessionID,
+							TriggeredAtOrigin:    "",
+							TriggeringEventType:  "",
+							TokenID:              "id1",
+							UserID:               "user-id",
+							OIDCSessionID:        "oidc-session-id1",
+							ClientID:             "client-id1",
+							BackChannelLogoutURI: "back-channel-logout-uri1",
+						},
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))),
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))),
+					).Return(nil)
+					q.EXPECT().Insert(gomock.Any(), &backchannel.LogoutRequest{
+						Aggregate:            sessionLogoutAgg,
+						SessionID:            sessionID,
+						TriggeredAtOrigin:    "",
+						TriggeringEventType:  "",
+						TokenID:              "id2",
+						UserID:               "user-id",
+						OIDCSessionID:        "oidc-session-id2",
+						ClientID:             "client-id2",
+						BackChannelLogoutURI: "back-channel-logout-uri2",
+					},
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))),
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))),
+					).Return(nil)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					return c
+				},
+				idGenerator: id_mock.NewIDGeneratorExpectIDs(t, "id1", "id2"),
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: testNow,
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate: sessionLogoutAgg,
+						SessionID: sessionID,
+					},
+				},
+			},
+			want: want{
+				err: nil,
+			},
+		},
+		{
+			name: "send logout request",
+			fields: fields{
+				es: expectEventstore(),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					c.EXPECT().BackChannelLogoutSent(gomock.Any(), sessionID, "oidc-session-id", instanceID).Return(nil)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					q.EXPECT().GetActiveSigningWebKey(gomock.Any()).Return(
+						&jose.JSONWebKey{
+							Key:       privateKey,
+							Algorithm: string(signingAlgorithm),
+							Use:       "sig",
+						}, nil)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					c.EXPECT().HandleMessage(gomock.Any()).DoAndReturn(
+						func(message channels.Message) error {
+							logoutMessage, ok := message.(*messages.Form)
+							if !ok {
+								ctrl.T.Errorf("unexpected message type: %T", message)
+							}
+							logoutToken, ok := logoutMessage.Serializable.(*LogoutTokenMessage)
+							if !ok {
+								ctrl.T.Errorf("unexpected serializable type: %T", logoutMessage.Serializable)
+							}
+							jws, err := jose.ParseSigned(logoutToken.LogoutToken, []jose.SignatureAlgorithm{jose.RS256})
+							require.NoError(t, err)
+							payload, err := jws.Verify(privateKey.Public())
+							require.NoError(t, err)
+							var claims oidc.LogoutTokenClaims
+							err = json.Unmarshal(payload, &claims)
+							require.NoError(t, err)
+							assert.Equal(t, "", claims.Issuer)
+							assert.Equal(t, "user-id", claims.Subject)
+							assert.Equal(t, "client-id", claims.Audience[0])
+							assert.WithinRange(t, claims.IssuedAt.AsTime(), testNow.Add(-2*time.Second), time.Now())
+							assert.WithinRange(t, claims.Expiration.AsTime(), testNow.Add(-time.Second).Add(time.Hour), time.Now().Add(time.Hour))
+							assert.Equal(t, "id1", claims.JWTID)
+							assert.NotNil(t, claims.Events["http://schemas.openid.net/event/backchannel-logout"])
+							assert.Equal(t, sessionID, claims.SessionID)
+
+							return nil
+						})
+					return c
+				},
+				idGenerator: id_mock.NewIDGeneratorExpectIDs(t),
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: testNow,
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate:            sessionLogoutAgg,
+						SessionID:            sessionID,
+						TriggeredAtOrigin:    "",
+						TriggeringEventType:  "",
+						TokenID:              "id1",
+						UserID:               "user-id",
+						OIDCSessionID:        "oidc-session-id",
+						ClientID:             "client-id",
+						BackChannelLogoutURI: "back-channel-logout-uri",
+					},
+				},
+			},
+			want: want{
+				err: nil,
+			},
+		},
+		{
+			name: "job creation failed",
+			fields: fields{
+				es: expectEventstore(
+					expectFilter(
+						eventFromEventPusher(
+							sessionlogout.NewBackChannelLogoutRegisteredEvent(
+								context.Background(),
+								sessionLogoutAgg,
+								"oidc-session-id1",
+								"user-id",
+								"client-id1",
+								"back-channel-logout-uri1",
+							),
+						),
+					),
+				),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					q.EXPECT().Insert(gomock.Any(),
+						gomock.Any(),
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))),
+						gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))),
+					).Return(assert.AnError)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					return c
+				},
+				idGenerator: id_mock.NewIDGeneratorExpectIDs(t, "id1"),
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: testNow,
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate: sessionLogoutAgg,
+						SessionID: sessionID,
+					},
+				},
+			},
+			want: want{
+				err: assert.AnError,
+			},
+		},
+
+		{
+			name: "send logout request failed",
+			fields: fields{
+				es: expectEventstore(),
+				queue: func(ctrl *gomock.Controller) Queue {
+					q := mock.NewMockQueue(ctrl)
+					return q
+				},
+				commands: func(ctrl *gomock.Controller) Commands {
+					c := mock.NewMockCommands(ctrl)
+					return c
+				},
+				queries: func(ctrl *gomock.Controller) Queries {
+					q := mock.NewMockQueries(ctrl)
+					q.EXPECT().GetActiveSigningWebKey(gomock.Any()).Return(
+						&jose.JSONWebKey{
+							Key:       privateKey,
+							Algorithm: string(signingAlgorithm),
+							Use:       "sig",
+						}, nil)
+					return q
+				},
+				channel: func(ctrl *gomock.Controller) channels.NotificationChannel {
+					c := channel_mock.NewMockNotificationChannel(ctrl)
+					c.EXPECT().HandleMessage(gomock.Any()).Return(assert.AnError)
+					return c
+				},
+				idGenerator: id_mock.NewIDGeneratorExpectIDs(t),
+			},
+			args: args{
+				job: &river.Job[*backchannel.LogoutRequest]{
+					JobRow: &rivertype.JobRow{
+						CreatedAt: testNow,
+					},
+					Args: &backchannel.LogoutRequest{
+						Aggregate:            sessionLogoutAgg,
+						SessionID:            sessionID,
+						TriggeredAtOrigin:    "",
+						TriggeringEventType:  "",
+						TokenID:              "id1",
+						UserID:               "user-id",
+						OIDCSessionID:        "oidc-session-id",
+						ClientID:             "client-id",
+						BackChannelLogoutURI: "back-channel-logout-uri",
+					},
+				},
+			},
+			want: want{
+				err: assert.AnError,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			err := newBackChannelLogoutWorker(
+				tt.fields.queries(ctrl),
+				tt.fields.commands(ctrl),
+				tt.fields.es(t),
+				tt.fields.queue(ctrl),
+				tt.fields.channel(ctrl),
+				tt.fields.idGenerator,
+				func() time.Time { return testNow },
+			).Work(
+				authz.WithInstanceID(context.Background(), instanceID),
+				tt.args.job,
+			)
+			assert.ErrorIs(t, err, tt.want.err)
+		})
+	}
+}
+
+func newBackChannelLogoutWorker(queries Queries, commands Commands, es *eventstore.Eventstore, queue Queue, channel channels.NotificationChannel, idGenerator id.Generator, testNow func() time.Time) *BackChannelLogoutWorker {
+	return &BackChannelLogoutWorker{
+		commands: commands,
+		queries: NewNotificationQueries(
+			queries,
+			es,
+			externalDomain,
+			externalPort,
+			externalSecure,
+			"",
+			nil,
+			nil,
+			nil,
+		),
+		eventstore: es,
+		queue:      queue,
+		channels: &notificationChannels{
+			Chain: *senders.ChainChannels(channel),
+		},
+		config: &BackChannelLogoutWorkerConfig{
+			Workers:             1,
+			TransactionDuration: 5 * time.Second,
+			MaxTtl:              5 * time.Minute,
+			MaxAttempts:         1,
+			TokenLifetime:       time.Hour,
+		},
+		now:         testNow,
+		idGenerator: idGenerator,
+	}
+}
+
+type expect func(mockRepository *es_repo_mock.MockRepository)
+
+func expectEventstore(expects ...expect) func(*testing.T) *eventstore.Eventstore {
+	return func(t *testing.T) *eventstore.Eventstore {
+		m := es_repo_mock.NewRepo(t)
+		for _, e := range expects {
+			e(m)
+		}
+		es := eventstore.NewEventstore(
+			&eventstore.Config{
+				Querier: m.MockQuerier,
+				Pusher:  m.MockPusher,
+			},
+		)
+		return es
+	}
+}
+
+func expectFilter(events ...eventstore.Event) expect {
+	return func(m *es_repo_mock.MockRepository) {
+		m.ExpectFilterEvents(events...)
+	}
+}
+
+func eventFromEventPusher(event eventstore.Command) *repository.Event {
+	data, _ := eventstore.EventData(event)
+	return &repository.Event{
+		InstanceID:    event.Aggregate().InstanceID,
+		ID:            "",
+		Seq:           0,
+		CreationDate:  time.Time{},
+		Typ:           event.Type(),
+		Data:          data,
+		EditorUser:    event.Creator(),
+		Version:       event.Aggregate().Version,
+		AggregateID:   event.Aggregate().ID,
+		AggregateType: event.Aggregate().Type,
+		ResourceOwner: sql.NullString{String: event.Aggregate().ResourceOwner, Valid: event.Aggregate().ResourceOwner != ""},
+		Constraints:   event.UniqueConstraints(),
+	}
+}
+
+var (
+	privateKey = func() *rsa.PrivateKey {
+		privateKey, _ := rsa.GenerateKey(rand.Reader, 2048)
+		return privateKey
+	}()
+	signingAlgorithm = jose.RS256
+)

internal/notification/handlers/commands.go

@@ -24,4 +24,5 @@ type Commands interface {
 	InviteCodeSent(ctx context.Context, orgID, userID string) error
 	UsageNotificationSent(ctx context.Context, dueEvent *quota.NotificationDueEvent) error
 	MilestonePushed(ctx context.Context, instanceID string, msType milestone.Type, endpoints []string) error
+	BackChannelLogoutSent(ctx context.Context, id, oidcSessionID, instanceID string) (err error)
 }

internal/notification/handlers/mock/commands.mock.go

@@ -43,6 +43,20 @@ func (m *MockCommands) EXPECT() *MockCommandsMockRecorder {
 	return m.recorder
 }
 
+// BackChannelLogoutSent mocks base method.
+func (m *MockCommands) BackChannelLogoutSent(arg0 context.Context, arg1, arg2, arg3 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "BackChannelLogoutSent", arg0, arg1, arg2, arg3)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// BackChannelLogoutSent indicates an expected call of BackChannelLogoutSent.
+func (mr *MockCommandsMockRecorder) BackChannelLogoutSent(arg0, arg1, arg2, arg3 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BackChannelLogoutSent", reflect.TypeOf((*MockCommands)(nil).BackChannelLogoutSent), arg0, arg1, arg2, arg3)
+}
+
 // HumanEmailVerificationCodeSent mocks base method.
 func (m *MockCommands) HumanEmailVerificationCodeSent(ctx context.Context, orgID, userID string) error {
 	m.ctrl.T.Helper()

internal/notification/projections.go

@@ -3,7 +3,6 @@ package notification
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/zitadel/logging"
 
@@ -12,6 +11,7 @@ import (
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
+	"github.com/zitadel/zitadel/internal/id"
 	"github.com/zitadel/zitadel/internal/notification/handlers"
 	_ "github.com/zitadel/zitadel/internal/notification/statik"
 	"github.com/zitadel/zitadel/internal/query"
@@ -27,6 +27,7 @@ func Register(
 	ctx context.Context,
 	userHandlerCustomConfig, quotaHandlerCustomConfig, telemetryHandlerCustomConfig, backChannelLogoutHandlerCustomConfig projection.CustomConfig,
 	notificationWorkerConfig handlers.WorkerConfig,
+	backChannelLogoutWorkerConfig *handlers.BackChannelLogoutWorkerConfig,
 	telemetryCfg handlers.TelemetryPusherConfig,
 	externalDomain string,
 	externalPort uint16,
@@ -35,8 +36,7 @@ func Register(
 	queries *query.Queries,
 	es *eventstore.Eventstore,
 	otpEmailTmpl, fileSystemPath string,
-	userEncryption, smtpEncryption, smsEncryption, keysEncryptionAlg crypto.EncryptionAlgorithm,
-	tokenLifetime time.Duration,
+	userEncryption, smtpEncryption, smsEncryption crypto.EncryptionAlgorithm,
 	queue *queue.Queue,
 ) {
 	if !notificationWorkerConfig.LegacyEnabled {
@@ -53,13 +53,11 @@ func Register(
 	projections = append(projections, handlers.NewBackChannelLogoutNotifier(
 		ctx,
 		projection.ApplyCustomConfig(backChannelLogoutHandlerCustomConfig),
-		commands,
 		q,
-		es,
-		keysEncryptionAlg,
-		c,
-		tokenLifetime,
+		queue,
+		backChannelLogoutWorkerConfig.MaxAttempts,
 	))
+	queue.AddWorkers(handlers.NewBackChannelLogoutWorker(commands, q, es, queue, c, backChannelLogoutWorkerConfig, id.SonyFlakeGenerator()))
 	if telemetryCfg.Enabled {
 		projections = append(projections, handlers.NewTelemetryPusher(ctx, telemetryCfg, projection.ApplyCustomConfig(telemetryHandlerCustomConfig), commands, q, c))
 	}