Add IAM replication drain worker

What does this MR do and why?

Adds the drain worker that propagates oauth_application changes from Rails to the IAM service. This implements Layer 3 (delayed microsync) of the IAM data propagation design: https://gitlab.com/gitlab-org/architecture/auth-architecture/design-doc/-/merge_requests/120

The iam_outbox capture half already exists: every create, update, or destroy of an Authn::OauthApplication writes an outbox row in the same transaction. Until now the drain seam was a no-op, so nothing consumed those rows. This MR adds the consumer.

How it works:

• Scheduling. Each outbox write schedules the worker with a ~1 minute delay, so any racing Rails writes have settled by the time it runs. The worker deduplicates on its arguments (entity_type, entity_id, event_type), including scheduled jobs, so a burst of same-entity mutations collapses into a single run.
• Delivery. Delivery goes through the real Authn::IamService::GrpcClient write RPCs. An upsert re-reads the current Rails state and sends the full OAuth client payload; a delete keys off the uid carried in the outbox payload and never reads Rails. Because IAM has no update RPC yet, an upsert is delivered as delete-then-create.
• Payload. The create request sends every field IAM needs: public, trusted, owner, grant_types, response_types, and timestamps. The owner string mirrors auth_helper#auth_app_owner_text so Rails and IAM render the same value.
• Failure handling. On an upstream failure the worker records the gRPC reason (a status-code label, never the upstream message) as l0_last_error, increments the attempt count, and re-raises so Sidekiq retries.

Scope this milestone:

• L0 target only. Only L0 is delivered; l0_* state is tracked and the l2_* columns are reserved for a follow-up.
• Everything is gated by the default-off iam_data_replication feature flag, checked both when writing the outbox row and when draining it.

Related to #602678 (closed)

References

• IAM data propagation design (ADR): https://gitlab.com/gitlab-org/architecture/auth-architecture/design-doc/-/merge_requests/120
• Rollout issue: #602678 (closed)

How to set up and validate locally

Enable the feature flag in the Rails console:

Feature.enable(:iam_data_replication)

• Have a working instance of iam-service up and running: (Configuration and Running the Service steps)
• Have IAM services configured in GitLab rails: gitlab-rails-local-setup.md

Happy flow (end to end)

1. Create an OAuth application under http://gdk.test:3000/oauth/applications, with a redirect URI of http://gdk.test:3000/callback and the api scope.

2. Wait ~1 minute for the drain worker to run, then confirm delivery in the Rails console:

   Authn::IamOutbox.last.l0_delivered_at # => present, for the correct entity_id

3. Initiate a consent flow against IAM, substituting your application's values:

   http://gdk.test:8084/oauth2/authorize?client_id=<CLIENT_ID>&redirect_uri=<REDIRECT_URI>&response_type=code&scope=<SCOPE>&state=<STATE>&code_challenge=<CODE_CHALLENGE>&code_challenge_method=S256

   Where:

   • <CLIENT_ID>: the application uid
   • <REDIRECT_URI>: http://gdk.test:3000/callback
   • <SCOPE>: api
   • <STATE>: any random string, e.g. my-random-state
   • <CODE_CHALLENGE>: a PKCE S256 challenge (paired with your verifier)

   Example:

   http://gdk.test:8084/oauth2/authorize?client_id=d563d74666270e5790a65708d9408adcea957a572f062aaa813b30c3d4217ed0&redirect_uri=http://gdk.test:3000/callback&response_type=code&scope=api&state=my-random-state&code_challenge=3lxayc3oq7fiA1sz4zQHkq4mdgw1PaQRGBLnTd8YxOE&code_challenge_method=S256

   The IAM consent page should render for the correct application, letting you authorize or cancel the request.

Manual check (Rails console)

1. Create or update an Authn::OauthApplication and drain it inline, bypassing the ~1 minute delay:

   app = Authn::OauthApplication.last
   Authn::IamReplication::DrainWorker.new.perform('oauth_application', app.id, 'upsert')

2. Confirm the outbox row is marked delivered:

   Authn::IamOutbox.where(entity_id: app.id).last.l0_delivered_at # => present

3. Destroy the application and drain the delete (keyed on the payload uid, no Rails read):

   app.destroy!
   Authn::IamReplication::DrainWorker.new.perform('oauth_application', app.id, 'delete')

4. With the flag disabled (Feature.disable(:iam_data_replication)), create an application and confirm no outbox row is written and the worker is a no-op.

Debugging

If a row is not delivered, inspect its per-target state:

row = Authn::IamOutbox.where(entity_id: app.id).last
row.attributes.slice('l0_delivered_at', 'l0_attempts', 'l0_last_error')

l0_last_error holds the gRPC reason (e.g. unavailable) for upstream failures, or the error class name for errors raised before the RPC.

Screenshots or screen recordings

Before	After

Query plan

Two operations were flagged by Danger, both on iam_outbox:

• the new l0_undelivered_lookup scope (app/models/authn/iam_outbox.rb:23)
• the update_all that marks rows delivered (app/workers/authn/iam_replication/drain_worker.rb:54)

The scope is never executed on its own, so the plan the worker actually produces at drain_worker.rb:34 (pending.last) is included as a third shape.

iam_outbox was created in 19.2 (db/migrate/20260618153007_create_iam_outbox.rb) behind the default-off iam_data_replication flag, so it is empty in production. Plans were captured on the gitlab-production-main clone seeded with a projected steady state of 105,003 rows:

• 100,000 delivered rows (l0_delivered_at set) on distinct entity_ids, excluded from the partial index
• 5,000 undelivered rows on distinct entity_ids, the concurrent backlog inside the partial index
• 3 undelivered rows on the target entity_id = 4242, with explicit ids so latest.id = 902000003

entity_id = 4242 also carries a delivered row from the first group, so the partial index has to separate pending rows from delivery history for the same entity.

The relevant index already exists, added with the table in db/migrate/20260618153007_create_iam_outbox.rb:28-30:

CREATE INDEX index_iam_outbox_l0_undelivered_lookup ON iam_outbox
  USING btree (entity_type, entity_id, event_type) WHERE (l0_delivered_at IS NULL);

1. l0_undelivered_lookup scope

SELECT — app/models/authn/iam_outbox.rb:23

https://postgres.ai/console/gitlab/gitlab-production-main/sessions/54631/commands/157528

SELECT "iam_outbox".* FROM "iam_outbox" WHERE "iam_outbox"."l0_delivered_at" IS NULL AND "iam_outbox"."entity_type" = 'oauth_application' AND "iam_outbox"."entity_id" = 4242 AND "iam_outbox"."event_type" = 0;

Query plan

Index Scan using index_iam_outbox_l0_undelivered_lookup on public.iam_outbox  (cost=0.29..3.31 rows=1 width=149) (actual time=0.057..0.059 rows=3 loops=1)
   Index Cond: ((iam_outbox.entity_type = 'oauth_application'::text) AND (iam_outbox.entity_id = 4242) AND (iam_outbox.event_type = 0))
   Buffers: shared hit=9
   I/O Timings: read=0.000 write=0.000
Settings: seq_page_cost = '4', effective_cache_size = '472585MB', jit = 'off', random_page_cost = '1.5', work_mem = '230MB'
Query ID: 8602181009555132159

Time: 0.959 ms
  - planning: 0.855 ms
  - execution: 0.104 ms
    - I/O read: 0.000 ms
    - I/O write: 0.000 ms

Shared buffers:
  - hits: 9 (~72.00 KiB) from the buffer pool
  - reads: 0 from the OS file cache, including disk I/O
  - dirtied: 0
  - writes: 0

2. Worker read: pending.last

SELECT — app/workers/authn/iam_replication/drain_worker.rb:34

https://postgres.ai/console/gitlab/gitlab-production-main/sessions/54631/commands/157529

SELECT "iam_outbox".* FROM "iam_outbox" WHERE "iam_outbox"."l0_delivered_at" IS NULL AND "iam_outbox"."entity_type" = 'oauth_application' AND "iam_outbox"."entity_id" = 4242 AND "iam_outbox"."event_type" = 0 ORDER BY "iam_outbox"."id" DESC LIMIT 1;

Query plan

Limit  (cost=3.32..3.32 rows=1 width=149) (actual time=0.059..0.060 rows=1 loops=1)
   Buffers: shared hit=12
   I/O Timings: read=0.000 write=0.000
   ->  Sort  (cost=3.32..3.32 rows=1 width=149) (actual time=0.058..0.059 rows=1 loops=1)
         Sort Key: iam_outbox.id DESC
         Sort Method: top-N heapsort  Memory: 25kB
         Buffers: shared hit=12
         I/O Timings: read=0.000 write=0.000
         ->  Index Scan using index_iam_outbox_l0_undelivered_lookup on public.iam_outbox  (cost=0.29..3.31 rows=1 width=149) (actual time=0.040..0.042 rows=3 loops=1)
               Index Cond: ((iam_outbox.entity_type = 'oauth_application'::text) AND (iam_outbox.entity_id = 4242) AND (iam_outbox.event_type = 0))
               Buffers: shared hit=9
               I/O Timings: read=0.000 write=0.000
Settings: work_mem = '230MB', seq_page_cost = '4', effective_cache_size = '472585MB', jit = 'off', random_page_cost = '1.5'
Query ID: 4215813893911831712

Time: 0.881 ms
  - planning: 0.779 ms
  - execution: 0.102 ms
    - I/O read: 0.000 ms
    - I/O write: 0.000 ms

Shared buffers:
  - hits: 12 (~96.00 KiB) from the buffer pool
  - reads: 0 from the OS file cache, including disk I/O
  - dirtied: 0
  - writes: 0

3. Mark delivered: update_all

UPDATE — app/workers/authn/iam_replication/drain_worker.rb:54

https://postgres.ai/console/gitlab/gitlab-production-main/sessions/54631/commands/157530

UPDATE "iam_outbox" SET "l0_delivered_at" = NOW(), "updated_at" = NOW() WHERE "iam_outbox"."l0_delivered_at" IS NULL AND "iam_outbox"."entity_type" = 'oauth_application' AND "iam_outbox"."entity_id" = 4242 AND "iam_outbox"."event_type" = 0 AND "iam_outbox"."id" <= 902000003;

Query plan

Update on public.iam_outbox  (cost=0.29..3.31 rows=0 width=0) (actual time=0.389..0.390 rows=0 loops=1)
   Buffers: shared hit=55
   WAL: records=12 fpi=4 bytes=23212
   I/O Timings: read=0.000 write=0.000
   ->  Index Scan using index_iam_outbox_l0_undelivered_lookup on public.iam_outbox  (cost=0.29..3.31 rows=1 width=22) (actual time=0.069..0.071 rows=3 loops=1)
         Index Cond: ((iam_outbox.entity_type = 'oauth_application'::text) AND (iam_outbox.entity_id = 4242) AND (iam_outbox.event_type = 0))
         Filter: ((iam_outbox.l0_delivered_at IS NULL) AND (iam_outbox.id <= 902000003))
         Buffers: shared hit=9
         I/O Timings: read=0.000 write=0.000
Settings: seq_page_cost = '4', effective_cache_size = '472585MB', jit = 'off', random_page_cost = '1.5', work_mem = '230MB'
Query ID: -884157393676940067

Time: 1.499 ms
  - planning: 1.017 ms
  - execution: 0.482 ms
    - I/O read: 0.000 ms
    - I/O write: 0.000 ms

Shared buffers:
  - hits: 55 (~440.00 KiB) from the buffer pool
  - reads: 0 from the OS file cache, including disk I/O
  - dirtied: 0
  - writes: 0

Rails emits literal timestamps in the SET list rather than NOW(), for example SET "l0_delivered_at" = '2026-08-12 15:53:56.804476', "updated_at" = '2026-08-12 15:53:56.804491'. NOW() was used when capturing the plan above. Both fold to a constant before execution and neither appears in the WHERE clause, so the plan is unaffected.

The error path at drain_worker.rb:121-122 (update_all on l0_last_error, then update_counters on l0_attempts) runs the same relation with the same WHERE clause; only the SET list differs. Both were captured from a logged run and match the statement above, so this plan applies to them too.

Conclusion

• All three shapes resolve through index_iam_outbox_l0_undelivered_lookup. No sequential scan appears in any plan, and no new index is required.
• Execution time is 0.104 ms, 0.102 ms and 0.482 ms respectively, against 9, 12 and 55 shared buffer hits with zero reads.

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.