Fall back to the stored registry location in Geo blob removal

What does this MR do and why?

When a Geo secondary misses a delete event, Geo::DestroyWorker removes the orphaned registry but passes no location. Once the model is gone, the secondary cannot find the blob, so the file stays on disk. This MR is the third step of the fix. It uses the location that !263322 (merged) records on each registry at sync time.

• BlobReplicatorStrategy#replicate_destroy still uses the delete event's location when there is one. It falls back to the registry's stored location only when the event data has no blob_path, which is the Geo::DestroyWorker case.
• The removal service also needs the uploader class. Upload registries store it. The other blob types derive it from the model's mounted uploader: file for most, external_diff for merge request diffs.
• The fallback is behind the geo_blob_removal_uses_registry_location feature flag, disabled by default. Registries synced before !263322 (merged) have no stored location, so they keep today's behavior until a later backfill.

References

Part of #281430

Database queries

1. ::Terraform::StateVersion.for_state_uuid_and_version(<uuid>,<version>).exists?

SELECT 1 AS one FROM "terraform_state_versions" INNER JOIN "terraform_states" ON "terraform_states"."id" = "terraform_state_versions"."terraform_state_id" WHERE "terraform_states"."uuid" = '<uuid>' AND "terraform_state_versions"."version" = <version> LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176933

⚠️ InitPlans 1 to 3 are there to pick an existing row. I don't have access to the production database and can't find valid uuid or version values myself.

The scope itself is the nested loop:

• an index scan on index_terraform_states_on_uuid, with rows=1
• then an index scan on index_terraform_state_versions_on_state_id_and_version, with rows=1

2. Upload.for_uploader_and_path(<uploader_class>, <path>).exists?

SELECT 1 AS one FROM "uploads" WHERE "uploads"."uploader" = '<uploader_class>' AND "uploads"."path" = '<path>' LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176939

or using the decomposed table:

SELECT 1 AS one FROM "project_uploads" WHERE "project_uploads"."uploader" = '<uploader_class>' AND "project_uploads"."path" = '<path>' LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176940

3. Packages::Nuget::Symbol.with_object_storage_key(<key>).exists?

SELECT 1 AS one FROM "packages_nuget_symbols" WHERE "packages_nuget_symbols"."object_storage_key" = <key> LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176941

4. ::Packages::Helm::MetadataCache.with_object_storage_key(<key>).exists?

SELECT 1 AS one FROM "packages_helm_metadata_caches" WHERE "packages_helm_metadata_caches"."object_storage_key" = <key> LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176942

⚠️ For the .with_object_storage_key query plans, the InitPlan only picks an existing object_storage_key, because I can't read production values. The scope itself is the top index scan on idx_pks_helm_metadata_caches_on_object_storage_key_project_id and index_packages_nuget_symbols_on_object_storage_key_project_id, returning one row. Its own cost is 6 buffers once the InitPlan's 4 are subtracted, and the timings are cold-cache reads.

5. model.primary_key_in(model_record_id).exists?

SELECT 1 AS one FROM "<table>" WHERE "<table>"."id" = <id> LIMIT 1

there are 38 different tables that could be used. I've used projects for the query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176943

6. LfsObject.for_oids(<oid>).exists?

SELECT 1 AS one FROM "lfs_objects" WHERE "lfs_objects"."oid" = '<oid>' LIMIT 1

Query plan: https://postgres.ai/console/gitlab/gitlab-production-main/sessions/72924/commands/176944

Screenshots or screen recordings

Not applicable: this MR has no UI changes.

How to set up and validate locally

1. In a GDK with a Geo secondary, enable the flag in a Rails console on the primary:

   Feature.enable(:geo_blob_removal_uses_registry_location)

2. In a Rails console on the secondary, pick a synced upload with a stored location and check that its file exists:

   registry = Geo::UploadRegistry.synced.where.not(blob_path: nil).last
   path = File.join(registry.uploader_class.constantize.root, registry.blob_path)
   File.exist?(path) # => true

3. On the primary, delete the upload row without callbacks, so no delete event is created:

   Upload.find(<registry.file_id>).delete

4. On the secondary, run the fallback that Geo::Secondary::RegistryConsistencyWorker would enqueue:

   Geo::DestroyWorker.new.perform('upload', registry.file_id)
   File.exist?(path) # => false

   With the flag disabled, the file stays, and geo.log shows "Unable to unlink file because file path is unknown. A file may be orphaned."

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.