Merge pull request #13421 from demarche-numerique/track_and_discard_purged_blob

Tech: corrige des problèmes avec le soft delete des blobs openstack

Author: LeSim

app/jobs/cron/purge_soft_deleted_blobs_job.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+class Cron::PurgeSoftDeletedBlobsJob < Cron::CronJob
+  self.schedule_expression = "every day at 01:00" # after PurgeUnattachedBlobsJob (00:30)
+
+  # Swift bulk delete accepts up to 10_000 objects per request; stay well under.
+  BULK_DELETE_LIMIT = 1000
+
+  def perform
+    return if ENV['PURGE_LATER_DELAY_IN_DAY'].blank?
+    return if ActiveStorage::Blob.service.name != :openstack
+
+    retention = Integer(ENV['PURGE_LATER_DELAY_IN_DAY']).days
+
+    ActiveStorage::Blob
+      .where(service_name: :openstack, soft_delete_at: ..retention.ago)
+      .in_batches(of: BULK_DELETE_LIMIT) { purge_batch(it.ids) }
+  end
+
+  private
+
+  # For a batch of expired blobs
+  # - add all the related variant blobs
+  # - delete all the corresponding files on the bucket
+  # - delete in db attachment / variant / blob
+  def purge_batch(parent_ids)
+    variant_record_ids = ActiveStorage::VariantRecord.where(blob_id: parent_ids).ids
+    image_attachments = ActiveStorage::Attachment
+      .where(record_type: 'ActiveStorage::VariantRecord', record_id: variant_record_ids)
+    blob_ids = image_attachments.pluck(:blob_id) + parent_ids
+
+    delete_files(blob_ids)
+
+    # delete_all bypasses callbacks, so we drop the rows ourselves, in FK order:
+    # image attachments -> variant records -> blobs (variants + parents).
+    image_attachments.delete_all
+    ActiveStorage::VariantRecord.where(id: variant_record_ids).delete_all
+    ActiveStorage::Blob.where(id: blob_ids).delete_all
+  end
+
+  def delete_files(blob_ids)
+    keys = ActiveStorage::Blob.where(id: blob_ids).pluck(:key)
+
+    service = ActiveStorage::Blob.service
+    client = service.send(:client)
+
+    keys.each_slice(BULK_DELETE_LIMIT) do |slice|
+      response = client.delete_multiple_objects(service.container, slice)
+      errors = response.body['Errors']
+      Sentry.capture_message("Bulk delete errors", extra: { errors: }) if errors.present?
+    end
+  end
+end

app/jobs/cron/purge_unattached_blobs_job.rb

@@ -16,6 +16,7 @@ def perform
     # caveats: the job needs to be run at least once a week to avoid missing blobs
     ActiveStorage::Blob
       .where(created_at: 1.week.ago..1.day.ago)
+      .where(soft_delete_at: nil)
       .unattached
       .select(:id, :service_name)
       .in_batches do |relation|

app/jobs/delayed_purge_job.rb

@@ -24,14 +24,8 @@ class DelayedPurgeJob < ApplicationJob
   discard_on ActiveJob::DeserializationError
   discard_on ActiveRecord::RecordNotFound
 
-  def self.openstack?
-    Rails.application.config.active_storage.service == :openstack
-  end
-
-  if openstack?
-    require 'fog/openstack'
-    discard_on Fog::OpenStack::Storage::NotFound
-  end
+  require 'fog/openstack'
+  discard_on Fog::OpenStack::Storage::NotFound
 
   delegate :service, :key, to: :blob
   delegate :container, to: :service
@@ -58,21 +52,40 @@ def soft_delete
     # We should replicate the same behavior here.
     # https://github.com/rails/rails/blob/ef88965e8a0c72496c210a5a0a48b85ec9a2ed17/activestorage/app/models/active_storage/blob.rb#L53-L55
     return if blob.attachments.exists?
-    excon_response = client.copy_object(container, key, container, key, { "Content-Type" => blob.content_type, 'X-Delete-At' => delay.to_s })
-    if excon_response.status != 201
-      Sentry.capture_message("Can't expire blob", extra: { key:, headers: })
-    else
-      service.delete_prefixed("variants/#{key}/") if blob.image?
-    end
+    return if !expire_file(blob)
+
+    blob.update_column(:soft_delete_at, Time.current)
+    soft_delete_variants if blob.image?
+  end
+
+  def soft_delete_variants
+    variant_blob_ids = ActiveStorage::Attachment
+      .where(record_type: 'ActiveStorage::VariantRecord', record_id: blob.variant_records.ids)
+      .pluck(:blob_id)
+
+    ActiveStorage::Blob.where(id: variant_blob_ids).find_each { expire_file(it) }
+  end
+
+  # Copy the blob's file onto itself with an X-Delete-At header so Swift expires
+  # it after the retention delay. Returns whether Swift accepted the request and
+  # reports to Sentry on failure.
+  def expire_file(blob)
+    headers = { "Content-Type" => blob.content_type, 'X-Delete-At' => delay.to_s }
+    return true if client.copy_object(container, blob.key, container, blob.key, headers).status == 201
+
+    Sentry.capture_message("Can't expire blob", extra: { key: blob.key, headers: })
+    false
   end
 
   def soft_delete_enabled?
-    DelayedPurgeJob.openstack? && delay.positive?
+    openstack? && delay.positive?
   rescue
     false
   end
 
+  def openstack? = service.name == :openstack
+
   def client
-    ActiveStorage::Blob.service.send(:client)
+    service.send(:client)
   end
 end

config/initializers/active_storage.rb

@@ -154,3 +154,31 @@ def publicize(url)
     end
   end
 end
+
+# fog-openstack 1.1.x builds the bulk-delete request body with `URI.encode`, which
+# Ruby removed in 3.0, so `delete_multiple_objects` raises NoMethodError on any
+# non-empty object list. `URI::DEFAULT_PARSER.escape` is the drop-in replacement:
+# same escaping rules, and it keeps the `container/object` slash literal (unlike
+# `ERB::Util.url_encode`, which would encode it and break the path).
+#
+# https://github.com/fog/fog-openstack/blob/v1.1.5/lib/fog/openstack/storage/requests/delete_multiple_objects.rb
+require 'fog/openstack'
+
+class Fog::OpenStack::Storage::Real
+  def delete_multiple_objects(container, object_names, options = {})
+    body = object_names.map do |name|
+      object_name = container ? "#{container}/#{name}" : name
+      URI::DEFAULT_PARSER.escape(object_name)
+    end.join("\n")
+
+    response = request({
+      expects: 200,
+      method: 'DELETE',
+      headers: options.merge('Content-Type' => 'text/plain', 'Accept' => 'application/json'),
+      body:,
+      query: { 'bulk-delete' => true },
+    }, false)
+    response.body = Fog::JSON.decode(response.body)
+    response
+  end
+end

db/migrate/20260703000000_add_soft_delete_at_to_active_storage_blobs.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class AddSoftDeleteAtToActiveStorageBlobs < ActiveRecord::Migration[7.2]
+  disable_ddl_transaction!
+
+  def change
+    add_column :active_storage_blobs, :soft_delete_at, :datetime, null: true, if_not_exists: true
+
+    add_index :active_storage_blobs, :soft_delete_at,
+              where: "soft_delete_at IS NOT NULL",
+              algorithm: :concurrently,
+              if_not_exists: true
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[8.0].define(version: 2026_06_25_153641) do
+ActiveRecord::Schema[8.0].define(version: 2026_07_03_000000) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "pg_buffercache"
   enable_extension "pg_stat_statements"
@@ -52,10 +52,12 @@
     t.text "metadata"
     t.jsonb "ocr"
     t.string "service_name", null: false
+    t.datetime "soft_delete_at"
     t.string "virus_scan_result"
     t.datetime "virus_scanned_at", precision: nil
     t.datetime "watermarked_at", precision: nil
     t.index ["key"], name: "index_active_storage_blobs_on_key", unique: true
+    t.index ["soft_delete_at"], name: "index_active_storage_blobs_on_soft_delete_at", where: "(soft_delete_at IS NOT NULL)"
     t.index ["virus_scan_result", "id"], name: "index_active_storage_blobs_on_pending_virus_scan", order: { id: :desc }
   end
 

spec/jobs/cron/purge_soft_deleted_blobs_job_spec.rb

@@ -0,0 +1,108 @@
+# frozen_string_literal: true
+
+RSpec.describe Cron::PurgeSoftDeletedBlobsJob, type: :job do
+  describe 'perform' do
+    # Blobs are created against the real (disk) service, then the job runs against
+    # a mocked OpenStack service — hence the helper, called from each example once
+    # every blob exists.
+    subject(:perform) do
+      service = double('openstack service', container: 'bucket', name: :openstack)
+      allow(service).to receive(:client).and_return(client) # reached via service.send(:client)
+      allow(ActiveStorage::Blob).to receive(:service).and_return(service)
+      described_class.perform_now
+    end
+
+    let(:client) { double('openstack client') }
+
+    before do
+      stub_const('ENV', ENV.to_hash.merge('PURGE_LATER_DELAY_IN_DAY' => '1'))
+      allow(client).to receive(:delete_multiple_objects).and_return(double(body: { 'Errors' => [] }))
+    end
+
+    # retention = PURGE_LATER_DELAY_IN_DAY = 1.day
+    let(:blob_old) do
+      ActiveStorage::Blob.create_and_upload!(io: StringIO.new("old"), filename: "old.txt", content_type: "text/plain").tap do |blob|
+        blob.update_columns(soft_delete_at: 2.days.ago, service_name: 'openstack')
+      end
+    end
+
+    let(:blob_recent) do
+      ActiveStorage::Blob.create_and_upload!(io: StringIO.new("recent"), filename: "recent.txt", content_type: "text/plain").tap do |blob|
+        blob.update_column(:soft_delete_at, 1.hour.ago)
+      end
+    end
+
+    let(:blob_never_soft_deleted) do
+      ActiveStorage::Blob.create_and_upload!(io: StringIO.new("kept"), filename: "kept.txt", content_type: "text/plain")
+    end
+
+    before do
+      blob_old
+      blob_recent
+      blob_never_soft_deleted
+    end
+
+    it 'destroys blobs soft-deleted long enough ago' do
+      expect { perform }.to change { ActiveStorage::Blob.exists?(id: blob_old.id) }.from(true).to(false)
+    end
+
+    it 'keeps recently soft-deleted and never-soft-deleted blobs' do
+      perform
+      expect(ActiveStorage::Blob.exists?(id: blob_recent.id)).to be(true)
+      expect(ActiveStorage::Blob.exists?(id: blob_never_soft_deleted.id)).to be(true)
+    end
+
+    it 'keeps blobs stored on another service' do
+      blob_other_service = ActiveStorage::Blob
+        .create_and_upload!(io: StringIO.new("other"), filename: "other.txt", content_type: "text/plain")
+        .tap { it.update_columns(soft_delete_at: 2.days.ago, service_name: 'test') }
+
+      perform
+
+      expect(ActiveStorage::Blob.exists?(id: blob_other_service.id)).to be(true)
+    end
+
+    it 'does nothing when the storage service is not OpenStack' do
+      allow(ActiveStorage::Blob).to receive(:service).and_return(double('disk service', name: :test))
+
+      expect { described_class.perform_now }.not_to change(ActiveStorage::Blob, :count)
+    end
+
+    it 'bulk-deletes the parent file (safety net against a missed X-Delete-At)' do
+      expect(client).to receive(:delete_multiple_objects)
+        .with('bucket', [blob_old.key])
+        .and_return(double(body: { 'Errors' => [] }))
+
+      perform
+    end
+
+    context 'when a soft-deleted image has variants' do
+      let!(:variant_blob) do
+        ActiveStorage::Blob.create_and_upload!(io: StringIO.new("variant"), filename: "v.png", content_type: "image/png")
+      end
+      let!(:variant_record) { ActiveStorage::VariantRecord.create!(blob: blob_old, variation_digest: "digest") }
+      let!(:image_attachment) { ActiveStorage::Attachment.create!(name: "image", record: variant_record, blob: variant_blob) }
+
+      it 'bulk-deletes both variant and parent files, then drops every row' do
+        expect(client).to receive(:delete_multiple_objects)
+          .with('bucket', match_array([variant_blob.key, blob_old.key]))
+          .and_return(double(body: { 'Errors' => [] }))
+
+        perform
+
+        expect(ActiveStorage::Blob.where(id: [blob_old.id, variant_blob.id])).not_to exist
+        expect(ActiveStorage::VariantRecord.where(id: variant_record.id)).not_to exist
+        expect(ActiveStorage::Attachment.where(id: image_attachment.id)).not_to exist
+      end
+
+      it 'reports per-object bulk delete errors to Sentry' do
+        errors = [["bucket/#{variant_blob.key}", "409 Conflict"]]
+        allow(client).to receive(:delete_multiple_objects).and_return(double(body: { 'Errors' => errors }))
+
+        expect(Sentry).to receive(:capture_message).with("Bulk delete errors", extra: { errors: })
+
+        perform
+      end
+    end
+  end
+end

spec/jobs/cron/purge_unattached_blobs_job_spec.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+RSpec.describe Cron::PurgeUnattachedBlobsJob, type: :job do
+  describe 'perform' do
+    subject { described_class.perform_now }
+
+    # unattached, within the created_at window
+    let(:blob) do
+      ActiveStorage::Blob.create_and_upload!(io: StringIO.new("data"), filename: "data.txt", content_type: "text/plain").tap do |b|
+        b.update_column(:created_at, 3.days.ago)
+      end
+    end
+
+    before { blob }
+
+    context 'when the blob has not been soft-deleted' do
+      it 'enqueues a purge' do
+        expect { subject }.to have_enqueued_job(DelayedPurgeJob)
+      end
+    end
+
+    context 'when the blob has already been soft-deleted' do
+      before { blob.update_column(:soft_delete_at, 1.hour.ago) }
+
+      it 'does not enqueue a purge' do
+        expect { subject }.not_to have_enqueued_job(DelayedPurgeJob)
+      end
+    end
+  end
+end