perf(audit_trail): index a record's history and its life boundaries

packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/sql/migrations.rb

@@ -62,6 +62,27 @@ module Migrations
               connection.add_index(table, %i[timestamp id], name: "#{table.split(".").last}_timestamp_id",
                                                             if_not_exists: true)
             end
+          },
+          {
+            # A record's history is read by (collection, record_id) in (timestamp, id) order. With only
+            # `record_id` and `(timestamp, id)` to choose from, Postgres either reads every row the id holds
+            # across all collections, or walks the whole table's (timestamp, id) index under a LIMIT: the life
+            # boundary lookup discarded 2.1M rows (486 ms) on a 2.1M-row table for a record created long ago.
+            name: '004-index-record-history',
+            up: lambda do |connection, table|
+              base = table.split('.').last
+              columns = %i[collection record_id timestamp id]
+              options = { if_not_exists: true }
+              options[:length] = { record_id: 255 } if connection.adapter_name.downcase.match?(/mysql|maria/)
+
+              connection.add_index(table, columns, name: "#{base}_record_history", **options)
+              next unless connection.supports_partial_index?
+
+              # The life boundary keeps a record's `create` and `delete` rows only: without this, a record
+              # updated 100k times is read whole to find its one `create`.
+              connection.add_index(table, columns, name: "#{base}_life_boundaries",
+                                                   where: "operation IN ('create', 'delete')", **options)
+            end
           }
         ].freeze
       end

packages/forest_admin_agent/lib/forest_admin_agent/audit_trail/store.rb

@@ -204,11 +204,12 @@ def authors(relation)
                 .uniq { |author| author[:user_id] }
       end
 
-      # Rows strictly before `row` in the trail's (timestamp, id) order.
+      # Rows strictly before `row` in the trail's (timestamp, id) order. A row value rather than the equivalent
+      # `timestamp < ? OR (timestamp = ? AND id < ?)`, which Postgres cannot turn into a range on the
+      # (collection, record_id, timestamp, id) index: on a 50k-row history the OR read all of it (7 ms), the row
+      # value a single entry (0.06 ms).
       def before_row_condition(row)
-        at = as_time(row.timestamp)
-
-        ['timestamp < ? OR (timestamp = ? AND id < ?)', at, at, row.id]
+        ['(timestamp, id) < (?, ?)', as_time(row.timestamp), row.id]
       end
 
       # Rows strictly past `row` in the history's own order, so a scan keyed on the last row it read neither

packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/sql/migrator_spec.rb

@@ -22,6 +22,7 @@ module Sql
             change_column: nil,
             remove_column: nil,
             column_exists?: true,
+            supports_partial_index?: true,
             select_values: ['001-create-audit-logs'],
             quote_schema_name: '"forest"',
             quote_table_name: '"forest.audit_logs_migration"',

packages/forest_admin_agent/spec/lib/forest_admin_agent/audit_trail/store_spec.rb

@@ -226,7 +226,8 @@ def timeline(**query)
         described_class.new(database: { adapter: 'sqlite3', database: db.path }).append(record)
 
         names = connection.select_values('SELECT name FROM audit_logs_migration ORDER BY name')
-        expect(names).to eq(%w[001-create-audit-logs 002-drop-previous-record-id 003-index-timestamp-id])
+        expect(names).to eq(%w[001-create-audit-logs 002-drop-previous-record-id 003-index-timestamp-id
+                               004-index-record-history])
       end
 
       # A table created by a release that still kept the column loses it on the next boot, rows intact.
@@ -253,7 +254,8 @@ def timeline(**query)
         expect(other.list_by_record(collection: 'accounts', record_id: '1').size).to eq(1)
         expect(connection.indexes('other_logs').map(&:name)).to include('other_logs_record_id')
         expect(connection.select_values('SELECT name FROM other_logs_migration'))
-          .to eq(%w[001-create-audit-logs 002-drop-previous-record-id 003-index-timestamp-id])
+          .to eq(%w[001-create-audit-logs 002-drop-previous-record-id 003-index-timestamp-id
+                    004-index-record-history])
       end
 
       # Every write sets it, so a row without one is a bug rather than a row quietly claiming to be done.
@@ -433,6 +435,20 @@ def boundary
           expect(boundary).to be_nil
         end
 
+        it 'counts a create as a boundary when a row sharing its instant was filed before it' do
+          store.append(record(operation: 'update', timestamp: '2026-01-02T00:00:00.000Z', correlation_key: 'earlier'))
+          store.append(record(operation: 'create', timestamp: '2026-01-02T00:00:00.000Z', correlation_key: 'retaken'))
+
+          expect(boundary).to eq('retaken')
+        end
+
+        it 'is nil for a create when the row sharing its instant was filed after it' do
+          store.append(record(operation: 'create', timestamp: '2026-01-02T00:00:00.000Z', correlation_key: 'first'))
+          store.append(record(operation: 'update', timestamp: '2026-01-02T00:00:00.000Z', correlation_key: 'same'))
+
+          expect(boundary).to be_nil
+        end
+
         it 'counts a delete as a boundary even with nothing before it, since it ends a life' do
           store.append(record(operation: 'delete', timestamp: '2026-01-02T00:00:00.000Z', correlation_key: 'freed'))
 
@@ -611,6 +627,34 @@ def matching_search(term)
         expect(connection.indexes('audit_logs').find { |index| index.name == 'audit_logs_timestamp_id' }.columns)
           .to eq(%w[timestamp id])
       end
+
+      it "indexes a record's history by collection, record_id, then timestamp and id" do
+        store.append(record)
+
+        history = connection.indexes('audit_logs').find { |index| index.name == 'audit_logs_record_history' }
+        expect([history.columns, history.where]).to eq([%w[collection record_id timestamp id], nil])
+      end
+
+      it 'indexes the same columns over create and delete rows only, for the life boundary' do
+        store.append(record)
+
+        boundaries = connection.indexes('audit_logs').find { |index| index.name == 'audit_logs_life_boundaries' }
+        expect([boundaries.columns, boundaries.where])
+          .to eq([%w[collection record_id timestamp id], "operation IN ('create', 'delete')"])
+      end
+
+      it 'skips the partial index on an adapter that cannot filter one' do
+        Sql::AuditConnectionBase.connect_to(adapter: 'sqlite3', database: db.path)
+        Sql::Migrations::ALL.first[:up].call(connection, 'audit_logs')
+        allow(connection).to receive(:supports_partial_index?).and_return(false)
+
+        Sql::Migrations::ALL.find { |migration| migration[:name] == '004-index-record-history' }[:up]
+                            .call(connection, 'audit_logs')
+
+        index_names = connection.indexes('audit_logs').map(&:name)
+        expect(index_names).to include('audit_logs_record_history')
+        expect(index_names).not_to include('audit_logs_life_boundaries')
+      end
     end
   end
 end