Class: Versions::SkippedColumnPurger

Inherits:
Object
  • Object
show all
Defined in:
app/services/versions/skipped_column_purger.rb

Overview

Removes audited columns from versions.object_changes that the owning models
already declare as skipped.

Models::Auditable.skip_audit_for was inert for its entire life: PaperTrail
resolves :skip / :ignore when has_paper_trail runs — inside included do,
which fires before the class-body call — so the declarations never reached it.
Six models believed they were excluding columns while the data accumulated
anyway. The declarations now take effect, but historical rows still carry the
payload, including Delivery#carrier_responses blobs reaching ~2 MB apiece.

Targets are derived from the models at runtime rather than hardcoded, so this
cleans exactly what the code currently claims to skip and needs no edit when a
declaration changes.

Space is NOT returned to the OS by this class

UPDATE and DELETE only mark tuples dead. Postgres reuses that space for future
writes but the files do not shrink. These partitions are append-only and have
never been vacuumed (zero dead tuples, autovacuum_count = 0), so this run is
the first thing ever to create garbage in them. To actually reclaim disk, follow
up per partition with pg_repack (installed; online, needs ~2x the partition
size free) or VACUUM FULL (ACCESS EXCLUSIVE — takes the table offline).

Examples:

Dry run over everything

Versions::SkippedColumnPurger.new(dry_run: true).call

Remediate the OAuth token rows only

Versions::SkippedColumnPurger.new(dry_run: false, only: %w[OauthCredential]).call

Defined Under Namespace

Classes: Config

Constant Summary collapse

DEFAULT_BATCH_SIZE =

Rows examined per statement. Each window is one index range scan on the
partition's (id, created_at) primary key.

50_000
DEFAULT_MAX_LAG_BYTES =

Abort if the standby falls further behind than this. A mass rewrite of an
append-only table generates far more WAL than this cluster normally sees.

256 * 1024 * 1024

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(dry_run: true, batch_size: DEFAULT_BATCH_SIZE, sleep_secs: 0.1, purge_empty: false, start_id: nil, max_lag_bytes: DEFAULT_MAX_LAG_BYTES, only: nil, allow_no_standby: false, logger: nil) ⇒ SkippedColumnPurger

Returns a new instance of SkippedColumnPurger.

Parameters:

  • dry_run (Boolean) (defaults to: true) —

    report only; issue no writes

  • batch_size (Integer) (defaults to: DEFAULT_BATCH_SIZE) —

    rows per statement

  • sleep_secs (Float) (defaults to: 0.1) —

    pause between batches, to bound WAL and IO pressure

  • purge_empty (Boolean) (defaults to: false) —

    also DELETE versions left with no changes at all

  • start_id (Integer, nil) (defaults to: nil) —

    resume point, for continuing an interrupted run

  • max_lag_bytes (Integer) (defaults to: DEFAULT_MAX_LAG_BYTES) —

    replication-lag ceiling before aborting

  • only (Array<String>, nil) (defaults to: nil) —

    restrict to these item_types

  • allow_no_standby (Boolean) (defaults to: false) —

    proceed even when no standby is streaming

  • logger (#info, nil) (defaults to: nil) —

    progress sink; prints to stdout when nil



59
60
61
62
63
64
65
66
67
 
# File 'app/services/versions/skipped_column_purger.rb', line 59

def initialize(dry_run: true, batch_size: DEFAULT_BATCH_SIZE, sleep_secs: 0.1,
               purge_empty: false, start_id: nil, max_lag_bytes: DEFAULT_MAX_LAG_BYTES,
               only: nil, allow_no_standby: false, logger: nil)
  @config = Config.new(dry_run:, batch_size:, sleep_secs:, purge_empty:, start_id:,
                       max_lag_bytes:, only: Array(only).presence, allow_no_standby:)
  validate_config!
  @logger = logger
  @stats  = { stripped: 0, emptied: 0, deleted: 0, batches: 0, last_id: nil }
end

Instance Attribute Details

#stats ⇒ Hash{Symbol => Integer, nil} (readonly)

Returns running totals for the current call.

Returns:

  • (Hash{Symbol => Integer, nil}) —

    running totals for the current call



48
49
50
 
# File 'app/services/versions/skipped_column_purger.rb', line 48

def stats
  @stats
end

Class Method Details

.targets ⇒ Hash{String => Array<String>}

Column names each model declares as skipped, keyed by item_type.

Reads both skipped_columns (what skip_audit_for recorded) and
paper_trail_options[:skip] (what PaperTrail actually honours, which is where
models like OauthCredential register directly). The union is "everything that
should never have been in the audit trail".

ALWAYS_IGNORED is excluded deliberately: it was always passed to
has_paper_trail correctly, and sampling both the oldest and newest live
partitions found zero rows carrying those keys.

Returns:

  • (Hash{String => Array<String>}) —

    item_type => sorted column names



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'app/services/versions/skipped_column_purger.rb', line 81

def self.targets
  Rails.application.eager_load!
  always = Models::Auditable::ALWAYS_IGNORED.map(&:to_s)

  ApplicationRecord.descendants.each_with_object({}) do |model, map|
    next if model.abstract_class? || !model.respond_to?(:paper_trail_options)

    declared  = Array(model.try(:skipped_columns)).map(&:to_s)
    effective = Array(model.paper_trail_options[:skip]).map(&:to_s)
    keys      = (declared | effective) - always
    next if keys.empty?

    map[model.name] = keys.sort
  end
end

Instance Method Details

#call ⇒ Hash{Symbol => Integer, nil}

Sweep every live partition, stripping (and optionally deleting) matched rows.

Returns:

  • (Hash{Symbol => Integer, nil}) —

    the accumulated #stats



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'app/services/versions/skipped_column_purger.rb', line 100

def call
  targets = self.class.targets
  targets = targets.slice(*@config.only) if @config.only

  if targets.empty?
    log 'No models declare skipped columns — nothing to do.'
    return stats
  end

  log(@config.dry_run ? '[DRY RUN — no rows will be modified]' : '[LIVE — rows will be rewritten]')
  targets.each { |item_type, keys| log "  #{item_type}: #{keys.join(' ')}" }

  partitions.each { |partition| process_partition(partition, targets) }

  log format('Done. stripped=%<stripped>d emptied=%<emptied>d deleted=%<deleted>d batches=%<batches>d',
             **stats.symbolize_keys)
  stats
end