diff --git a/app/jobs/cron/purge_soft_deleted_blobs_job.rb b/app/jobs/cron/purge_soft_deleted_blobs_job.rb new file mode 100644 index 00000000000..e851f687588 --- /dev/null +++ b/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 diff --git a/app/jobs/cron/purge_unattached_blobs_job.rb b/app/jobs/cron/purge_unattached_blobs_job.rb index 5a324b8b8c5..74fcfd657ea 100644 --- a/app/jobs/cron/purge_unattached_blobs_job.rb +++ b/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| diff --git a/app/jobs/delayed_purge_job.rb b/app/jobs/delayed_purge_job.rb index 4208edf93ef..7c483bfa598 100644 --- a/app/jobs/delayed_purge_job.rb +++ b/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 diff --git a/config/initializers/active_storage.rb b/config/initializers/active_storage.rb index 56c28d0c832..eee20939266 100644 --- a/config/initializers/active_storage.rb +++ b/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 diff --git a/db/migrate/20260703000000_add_soft_delete_at_to_active_storage_blobs.rb b/db/migrate/20260703000000_add_soft_delete_at_to_active_storage_blobs.rb new file mode 100644 index 00000000000..d52ef9e5167 --- /dev/null +++ b/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 diff --git a/db/schema.rb b/db/schema.rb index b1fe802dd30..531e55e41d7 100644 --- a/db/schema.rb +++ b/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 diff --git a/spec/jobs/cron/purge_soft_deleted_blobs_job_spec.rb b/spec/jobs/cron/purge_soft_deleted_blobs_job_spec.rb new file mode 100644 index 00000000000..04d72df1db8 --- /dev/null +++ b/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 diff --git a/spec/jobs/cron/purge_unattached_blobs_job_spec.rb b/spec/jobs/cron/purge_unattached_blobs_job_spec.rb new file mode 100644 index 00000000000..bcf36272e2c --- /dev/null +++ b/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 diff --git a/spec/jobs/delayed_purge_job_spec.rb b/spec/jobs/delayed_purge_job_spec.rb index d82b5ebc172..ae53ba9e178 100644 --- a/spec/jobs/delayed_purge_job_spec.rb +++ b/spec/jobs/delayed_purge_job_spec.rb @@ -19,13 +19,12 @@ context 'emit request instead of destroying it' do let(:container) { "bucket" } let(:client) { double("client") } - let(:double_service) { double(container:) } + let(:double_service) { double(name: :openstack, container:) } let(:cloned_dossier) { dossier.clone } before do allow_any_instance_of(ActiveStorage::Blob).to receive(:service).and_return(double_service) allow_any_instance_of(DelayedPurgeJob).to receive(:client).and_return(client) - allow(described_class).to receive(:openstack?).and_return(true) end it 'with attachments' do @@ -41,6 +40,16 @@ .and_return(double(status: 201)) subject perform_enqueued_jobs + expect(blob.reload.soft_delete_at).to be_present + end + + it 'does not mark soft_delete_at when the copy fails' do + dossier.champ_data.first.piece_justificative_file.first.delete + expect(client).to receive(:copy_object).and_return(double(status: 500)) + expect(Sentry).to receive(:capture_message) + subject + perform_enqueued_jobs + expect(blob.reload.soft_delete_at).to be_nil end it 'with cloned dossier' do @@ -56,6 +65,54 @@ end end + context 'when a soft-deleted image has variants' do + let(:container) { "bucket" } + let(:client) { double("client") } + + # Blobs are created against the real service, then the OpenStack service is + # mocked; the parent copy_object is stubbed to succeed. + def setup_image_with_variant + image_blob = ActiveStorage::Blob.create_and_upload!(io: StringIO.new("img"), filename: "i.png", content_type: "image/png") + variant_blob = ActiveStorage::Blob.create_and_upload!(io: StringIO.new("variant"), filename: "v.png", content_type: "image/png") + variant_record = ActiveStorage::VariantRecord.create!(blob: image_blob, variation_digest: "digest") + image_attachment = ActiveStorage::Attachment.create!(name: "image", record: variant_record, blob: variant_blob) + + allow_any_instance_of(ActiveStorage::Blob).to receive(:service).and_return(double(name: :openstack, container:)) + job = described_class.new(image_blob) + allow(job).to receive(:client).and_return(client) + allow(client).to receive(:copy_object).and_return(double(status: 201)) + + { job:, image_blob:, variant_blob:, variant_record:, image_attachment: } + end + + it 'marks the variant files for expiration and keeps their rows for the cron' do + ctx = setup_image_with_variant + expect(client).to receive(:copy_object) + .with(container, ctx[:variant_blob].key, container, ctx[:variant_blob].key, hash_including('X-Delete-At')) + .and_return(double(status: 201)) + + ctx[:job].perform_now + + expect(ActiveStorage::Blob.where(id: ctx[:variant_blob].id)).to exist + expect(ActiveStorage::VariantRecord.where(id: ctx[:variant_record].id)).to exist + expect(ActiveStorage::Attachment.where(id: ctx[:image_attachment].id)).to exist + expect(ActiveStorage::Blob.where(id: ctx[:image_blob].id)).to exist # parent soft-deleted + end + + it 'reports to Sentry when a variant copy fails, keeping the rows' do + ctx = setup_image_with_variant + allow(client).to receive(:copy_object) + .with(container, ctx[:variant_blob].key, container, ctx[:variant_blob].key, anything) + .and_return(double(status: 500)) + expect(Sentry).to receive(:capture_message).with("Can't expire blob", extra: hash_including(key: ctx[:variant_blob].key)) + + ctx[:job].perform_now + + expect(ActiveStorage::VariantRecord.where(id: ctx[:variant_record].id)).to exist + expect(ActiveStorage::Blob.where(id: ctx[:variant_blob].id)).to exist + end + end + context 'when destroying an instance' do it 'uses our custom job' do expect { dossier.destroy }.to have_enqueued_job(DelayedPurgeJob) @@ -63,6 +120,19 @@ end end + context 'when the blob is stored on a non-OpenStack service (e.g. S3)' do + let(:s3_service) { double(name: :amazon) } + + before do + allow_any_instance_of(ActiveStorage::Blob).to receive(:service).and_return(s3_service) + end + + it 'purges the blob directly instead of soft-deleting it' do + expect(blob).to receive(:purge) + subject + end + end + context 'error handling' do context 'Excon::Error::RequestEntityTooLarge' do let(:error) { Excon::Error::RequestEntityTooLarge.new('Request Entity Too Large') } diff --git a/spec/lib/active_storage/fog_openstack_bulk_delete_spec.rb b/spec/lib/active_storage/fog_openstack_bulk_delete_spec.rb new file mode 100644 index 00000000000..02f6ecca059 --- /dev/null +++ b/spec/lib/active_storage/fog_openstack_bulk_delete_spec.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +require 'rails_helper' + +# Guards the monkeypatch in config/initializers/active_storage.rb that repairs +# fog-openstack's `delete_multiple_objects` (broken by the removal of URI.encode +# in Ruby 3.0). +describe Fog::OpenStack::Storage::Real, '#delete_multiple_objects' do + subject(:real) { described_class.allocate } + + let(:captured) { {} } + + before do + allow(real).to receive(:request) do |params, _| + captured.merge!(params) + Struct.new(:body).new('{"Number Deleted": 2, "Errors": []}') + end + end + + it 'does not raise (URI.encode is gone) and escapes each path, keeping the slash literal' do + expect { real.delete_multiple_objects('bucket', ['key one', 'variants/abc/def']) } + .not_to raise_error + + expect(captured[:body]).to eq("bucket/key%20one\nbucket/variants/abc/def") + expect(captured[:query]).to eq('bulk-delete' => true) + expect(captured[:method]).to eq('DELETE') + end + + it 'decodes the JSON response body' do + response = real.delete_multiple_objects('bucket', ['key']) + + expect(response.body).to eq('Number Deleted' => 2, 'Errors' => []) + end + + # Canaries: the monkeypatch is only needed while BOTH remain true — Ruby lacks + # URI.encode AND fog-openstack still calls it. When either stops holding, the + # matching test fails: that is the signal to delete the patch and these tests. + describe 'monkeypatch necessity' do + it 'URI.encode is still undefined in this Ruby' do + expect { URI.encode('x') }.to raise_error(NoMethodError) + end + + it 'fog-openstack still ships the broken URI.encode call' do + source_file = File.join( + Gem.loaded_specs.fetch('fog-openstack').gem_dir, + 'lib/fog/openstack/storage/requests/delete_multiple_objects.rb' + ) + + expect(File.read(source_file)).to include('URI.encode') + end + end +end