Module: ActiveRecord::Materialized::WriteOutbox

Defined in:
lib/activerecord/materialized/write_outbox.rb

Overview

The trigger/outbox change source — the completeness layer for writes that bypass both ActiveRecord callbacks and the ingestion API: raw SQL from a console, a bulk backfill, another service writing the shared database. Database triggers on a dependency table capture the changed partition keys into a durable outbox table, and WriteOutbox.drain! relays them through the ingestion API (ingest_change) so the view is maintained exactly as it would be for an ActiveRecord write. See docs/out-of-band-writes.md.

Capture is scoped: each trigger records only the configured key columns (the GROUP BY keys), as JSON, in key_before (the OLD image) and/or key_after (the NEW image). That is precisely what ActiveRecord::Materialized::WriteChange.from_descriptor needs to scope maintenance to the affected partition(s) — and, for an update that moves a row between partitions, to maintain both the old and the new.

This is deliberately not the default change source: triggers are DDL a host app opts into (via the activerecord_materialized:outbox generator) for the tables where out-of-band writes actually happen. Drift detection (#62) + reconcile_stale! remain the time-bounded backstop.

Defined Under Namespace

Modules: Triggers

Class Method Summary collapse

Class Method Details

.drain!(limit: nil) ⇒ Integer

Relay pending outbox rows through the ingestion API, in write order, deleting each after it is successfully relayed. Run from a poller, cron, or background job. Properties:

  • Bounded memory. Rows are relayed in batches of DRAIN_BATCH_SIZE, so even a huge backlog (the bulk-backfill case) drains without loading the whole outbox at once.
  • At-least-once. A row is deleted only after its relay succeeds, so a crash mid-drain re-relays the not-yet-deleted rows next pass — safe, since ActiveRecord::Materialized.ingest_change is convergent (scoped recompute is idempotent).
  • Per-row isolation. A relay that raises (e.g. a view whose scoped recompute fails) leaves only that row in the outbox for retry and is skipped for the rest of this pass, so one poison row can't block the writes queued behind it (mirrors Reconciler's per-view isolation).

A no-op (returns 0) before any triggers are installed, so a scheduled drain is safe to run even if the outbox migration hasn't been applied yet.

Parameters:

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

    max rows to attempt this pass (nil = drain everything reachable)

Returns:

  • (Integer) —

    the number of outbox rows successfully relayed and deleted



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/activerecord/materialized/write_outbox.rb', line 93

def drain!(limit: nil)
  return 0 unless WriteOutboxRecord.connection.data_source_exists?(table_name)

  drained = 0
  attempted = 0
  failed_ids = []
  loop do
    take = limit ? [limit - attempted, DRAIN_BATCH_SIZE].min : DRAIN_BATCH_SIZE
    break if take <= 0

    rows = pending_rows(failed_ids, take)
    break if rows.empty?

    attempted += rows.size
    relayed_count, failed = relay_and_delete(rows)
    drained += relayed_count
    failed_ids.concat(failed)
  end
  drained
end

.ensure_table!(connection = ActiveRecord::Base.connection) ⇒ void

This method returns an undefined value.

Lazily provision the outbox table, mirroring Metadata::Schema. Idempotent.

Parameters:

  • connection (ActiveRecord::ConnectionAdapters::AbstractAdapter) (defaults to: ActiveRecord::Base.connection) 


34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/activerecord/materialized/write_outbox.rb', line 34

def ensure_table!(connection = ActiveRecord::Base.connection)
  return if connection.data_source_exists?(table_name)

  connection.create_table(table_name) do |t|
    t.string :source_table, null: false # the dependency table that was written
    t.string :operation, null: false # one of WriteChange::OPERATIONS ("create"/"update"/"destroy")
    t.text :key_before # JSON of the OLD-image key columns (destroy/update); NULL on create
    t.text :key_after # JSON of the NEW-image key columns (create/update); NULL on destroy
    t.datetime :created_at, null: false
  end
  WriteOutboxRecord.reset_column_information # mirror Metadata::Schema: drop any stale column cache
end

.install_triggers!(table, key_columns:, connection: ActiveRecord::Base.connection) ⇒ void

This method returns an undefined value.

Install AFTER INSERT/UPDATE/DELETE triggers on table that append the key_columns values to the outbox on every write — including raw SQL that never touches ActiveRecord. Idempotent: re-installing drops and recreates. Portable across the gem's adapters (SQLite/MySQL/Postgres); the generated migration runs this so the correct dialect is emitted at migrate time.

Parameters:

  • table (String, Symbol) —

    the dependency table to watch

  • key_columns (Array<String, Symbol>) —

    the GROUP BY key columns to capture (empty for an un-grouped/global view — every write then relays an empty image, widening to a full recompute)

  • connection (ActiveRecord::ConnectionAdapters::AbstractAdapter) (defaults to: ActiveRecord::Base.connection) 


57
58
59
60
 
# File 'lib/activerecord/materialized/write_outbox.rb', line 57

def install_triggers!(table, key_columns:, connection: ActiveRecord::Base.connection)
  ensure_table!(connection)
  Triggers.for(connection).install!(connection, table.to_s, Array(key_columns).map(&:to_s))
end

.table_name ⇒ String

Returns the outbox table name (configurable via config.write_outbox_table_name).

Returns:

  • (String) —

    the outbox table name (configurable via config.write_outbox_table_name)



26
27
28
 
# File 'lib/activerecord/materialized/write_outbox.rb', line 26

def table_name
  ActiveRecord::Materialized.configuration.write_outbox_table_name
end

.uninstall_triggers!(table, connection: ActiveRecord::Base.connection) ⇒ void

This method returns an undefined value.

Remove the triggers (and, on Postgres, the trigger function) installed by install_triggers!.

Parameters:

  • table (String, Symbol)
  • connection (ActiveRecord::ConnectionAdapters::AbstractAdapter) (defaults to: ActiveRecord::Base.connection) 


67
68
69
 
# File 'lib/activerecord/materialized/write_outbox.rb', line 67

def uninstall_triggers!(table, connection: ActiveRecord::Base.connection)
  Triggers.for(connection).uninstall!(connection, table.to_s)
end