Use batched iteration for NamespaceIndexIntegrityWorker

What does this MR do and why?

Summary

Previously, when checking search index integrity for a namespace (a group of projects), the worker would immediately loop through all child groups and projects at once, which could be overwhelming for large namespaces.

This update introduces batched, resumable processing: instead of doing everything in one go, the worker now processes descendants in chunks of 500 at a time. If it schedules more than 2,000 items in a single run, it pauses and reschedules itself to continue from where it left off (using a "cursor" to track its position). This prevents the worker from overloading the system when dealing with very large namespace trees.

Additionally, the deduplication behavior was updated so that if a duplicate job is detected, it will reschedule itself once rather than simply being dropped — ensuring no work is silently skipped.

Query Plans

1. Namespace CTE Query

Click to expand for query

WITH RECURSIVE "result" AS (
    -- Base case: Start from namespace 9970
    (SELECT
      9970::bigint AS current_id,
      ARRAY[9970]::bigint[] AS depth,
      ARRAY[9970]::bigint[] AS ids,
      1::bigint AS count,
      0::bigint AS index
     FROM (VALUES (1)) AS initializer_row
     WHERE (EXISTS (SELECT 1 FROM "namespaces" WHERE "namespaces"."id" = 9970)))

    UNION ALL

    -- Recursive case: Three branches (walk_down, next_elements, up_one_level)
    (WITH "cte" AS MATERIALIZED (SELECT result.* FROM result)
     SELECT current_id, depth, ids, count, index
     FROM (
       -- Branch 1: Walk down to first child
       (SELECT
          "namespaces"."id" AS current_id,
          cte.depth || "namespaces".id::bigint AS depth,
          cte.ids || "namespaces".id::bigint AS ids,
          cte.count + 1 AS count,
          1::bigint AS index
        FROM cte,
        LATERAL (
          SELECT "namespaces"."id"
          FROM "namespaces"
          WHERE (parent_id = cte.current_id)
          ORDER BY "namespaces"."id" ASC
          LIMIT 1
        ) "namespaces")

       UNION ALL

       -- Branch 2: Get next sibling at same level
       (SELECT
          "namespaces"."id" AS current_id,
          cte.depth[:array_length(cte.depth, 1) - 1] || "namespaces".id::bigint AS depth,
          cte.ids || "namespaces".id::bigint AS ids,
          cte.count + 1 AS count,
          2::bigint AS index
        FROM cte,
        LATERAL (
          SELECT "namespaces"."id"
          FROM "namespaces"
          WHERE ("namespaces".parent_id = cte.depth[array_length(cte.depth, 1) - 1])
            AND ("namespaces".id > cte.depth[array_length(cte.depth, 1)])
          ORDER BY "namespaces"."id" ASC
          LIMIT 1
        ) "namespaces")

       UNION ALL

       -- Branch 3: Move up one level when no more siblings
       (SELECT
          cte.current_id AS current_id,
          cte.depth[:array_length(cte.depth, 1) - 1] AS depth,
          cte.ids AS ids,
          cte.count + 1 AS count,
          3::bigint AS index
        FROM cte
        WHERE (cte.depth <> '{}')
        LIMIT 1)
     ) namespaces
     ORDER BY "namespaces"."index" ASC
     LIMIT 1)
  )
  SELECT current_id, depth, ids, count, index
  FROM "result" AS "namespaces"
  LIMIT 501;

Plan Link: https://console.postgres.ai/gitlab/gitlab-production-main/sessions/53498/commands/155557

2. Filtering Project Namespaces

Click to expand

  SELECT "namespaces"."id"
  FROM "namespaces"
  WHERE "namespaces"."type" = 'Project'
    AND "namespaces"."id" IN (
      -- namespace_ids here
    );

Plan Link: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/53498/commands/155558

3. Find Projects by ProjectNamespace

Click to expand

  SELECT "namespaces"."id"
  FROM "namespaces"
  WHERE "namespaces"."type" = 'Project'
    AND "namespaces"."id" IN (
      -- namespace_ids here
    );

Plan Link: https://console.postgres.ai/gitlab/projects/gitlab-production-main/sessions/53498/commands/155559

Screenshots or screen recordings

Before	After

How to set up and validate locally

1. Can be validated by running the spec added.

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.