Fix squash query to handle concurrent updates

Riddlerrr · Riddlerrr · commit 22191a09dcc3 · 2026-06-30T11:57:13.000+02:00
Widen the validity lookup from `= _now` to `&gt;= _now` so that the squash logic finds history rows written by a concurrent transaction whose `lower(validity)` may be slightly ahead of the current transaction's start time.
Also, added `upper_inf(validity)` filter for find only open-ended ranges. We can have only one open range by system design.

After the new `FOR UPDATE` lock serializes concurrent updates, the second transaction's `_now` is set before it acquires the lock and can be earlier than the `lower(validity)` written by the first transaction. The strict equality missed that row, falling through to the `ELSE` branch and triggering a `PG::ExclusionViolation` or
`RangeError` on overlapping validity ranges.
diff --git a/README.sql b/README.sql
@@ -111,10 +111,12 @@ create or replace function chronomodel_countries_update() returns trigger as $$
       return new;
     end if;
 
+    perform 1 from only temporal.countries where id = old.id for update;
+
     _now := timezone('UTC', now());
     _hid := null;
 
-    select hid into _hid from history.countries where id = old.id and lower(validity) = _now;
+    select hid into _hid from history.countries where id = old.id and upper_inf(validity) and lower(validity) >= _now limit 1;
 
     if _hid is not null then
       update history.countries set ( name, updated_at ) = ( new.name, new.updated_at ) where hid = _hid;
diff --git a/lib/chrono_model/adapter/ddl.rb b/lib/chrono_model/adapter/ddl.rb
@@ -154,10 +154,12 @@ def chrono_create_UPDATE_trigger(table, pk, current, history, fields, values, op
                         RETURN NEW;
                     END IF;
 
+                    PERFORM 1 FROM ONLY #{current} WHERE #{pk} = OLD.#{pk} FOR UPDATE;
+
                     _now := timezone('UTC', now());
                     _hid := NULL;
 
-                    #{"SELECT hid INTO _hid FROM #{history} WHERE #{pk} = OLD.#{pk} AND lower(validity) = _now;" unless ENV['CHRONOMODEL_NO_SQUASH']}
+                    #{"SELECT hid INTO _hid FROM #{history} WHERE #{pk} = OLD.#{pk} AND upper_inf(validity) AND lower(validity) >= _now LIMIT 1;" unless ENV['CHRONOMODEL_NO_SQUASH']}
 
                     IF _hid IS NOT NULL THEN
                         UPDATE #{history} SET (#{fields}) = (#{values}) WHERE hid = _hid;
diff --git a/spec/chrono_model/history_models_spec.rb b/spec/chrono_model/history_models_spec.rb
@@ -20,6 +20,9 @@
       expected['sub_bars'] = SubBar::History  if defined?(SubBar::History)
       expected['sub_sub_bars'] = SubSubBar::History if defined?(SubSubBar::History)
 
+      expected['overlappers'] = Overlapper::History if defined?(Overlapper::History)
+      expected['overlap_bros'] = OverlapBro::History if defined?(OverlapBro::History)
+
       # default_scope_spec
       expected['defoos']   = Defoo::History   if defined?(Defoo::History)
 
diff --git a/spec/chrono_model/time_machine/concurrency_spec.rb b/spec/chrono_model/time_machine/concurrency_spec.rb
@@ -0,0 +1,124 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'support/time_machine/structure'
+
+RSpec.describe ChronoModel::TimeMachine do
+  include ChronoTest::TimeMachine::Helpers
+
+  describe 'concurrent updates' do
+    it 'handles concurrent updates without PG::ExclusionViolation' do
+      foo = Foo.create!(name: 'concurrency-test')
+
+      threads_count = 4
+      iterations = 10
+      errors = []
+      mutex = Mutex.new
+
+      threads = Array.new(threads_count) do |i|
+        Thread.new do
+          iterations.times do |j|
+            Foo.transaction do
+              record = Foo.find(foo.id)
+              record.update!(name: "iteration-#{i}-#{j}")
+            end
+          rescue StandardError => e
+            mutex.synchronize do
+              errors << "Thread #{i} Iteration #{j}: #{e.class} - #{e.message}"
+            end
+          end
+        end
+      end
+
+      threads.each(&:join)
+
+      expect(errors).to be_empty
+    ensure
+      # Clean up: delete the record and its history
+      if foo
+        foo.destroy
+        Foo::History.where(id: foo.id).delete_all
+      end
+    end
+
+    it 'keeps the open history row aligned with the current row when an older transaction commits last' do
+      foo = Foo.create!(name: 'orig')
+      errors = []
+      mutex = Mutex.new
+      started = Queue.new
+      proceed = Queue.new
+      stale = nil
+      stale_started = false
+
+      begin
+        stale = Thread.new do
+          Foo.connection_pool.with_connection do
+            Foo.transaction do
+              Foo.connection.execute("SELECT timezone('UTC', now())")
+              stale_started = true
+              started << true
+              proceed.pop
+              Foo.find(foo.id).update!(name: 'stale')
+            end
+          end
+        rescue StandardError => e
+          started << true unless stale_started
+          mutex.synchronize { errors << e }
+        end
+
+        started.pop
+        sleep(0.05)
+        Foo.find(foo.id).update!(name: 't1')
+        sleep(0.05)
+        Foo.find(foo.id).update!(name: 't2')
+
+        proceed << true
+        stale.join
+
+        expect(errors).to be_empty
+        expect(foo.reload.name).to eq('stale')
+        expect(Foo::History.where(id: foo.id).where('upper_inf(validity)').pick(:name)).to eq(foo.name)
+      ensure
+        proceed << true if proceed && stale&.alive?
+        stale&.join
+
+        if foo
+          Foo.where(id: foo.id).delete_all
+          Foo::History.where(id: foo.id).delete_all
+        end
+      end
+    end
+  end
+
+  describe 'concurrent deletions' do
+    it 'handles concurrent deletions without error' do
+      threads_count = 4
+      foo_ids = Array.new(threads_count) { |i| Foo.create!(name: "test-delete-#{i}") }.map(&:id)
+      errors = []
+      mutex = Mutex.new
+
+      threads = Array.new(threads_count) do |i|
+        Thread.new do
+          Foo.where(id: foo_ids[i]).delete_all
+        rescue StandardError => e
+          mutex.synchronize do
+            errors << "Thread #{i}: #{e.class} - #{e.message}"
+          end
+        end
+      end
+
+      threads.each(&:join)
+
+      expect(errors).to be_empty
+      foo_ids.each do |id|
+        expect(Foo.exists?(id)).to be(false)
+      end
+    ensure
+      # Clean up any remaining records and history
+      foo_ids&.each do |id|
+        Foo::History.where(id: id).delete_all
+        Foo.where(id: id).delete_all
+      end
+    end
+  end
+end
diff --git a/spec/chrono_model/time_machine/transaction_timestamp_spec.rb b/spec/chrono_model/time_machine/transaction_timestamp_spec.rb
@@ -0,0 +1,147 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'support/time_machine/structure'
+
+RSpec.describe ChronoModel::TimeMachine do
+  include ChronoTest::TimeMachine::Helpers
+
+  after do
+    Overlapper::History.delete_all
+    Overlapper.delete_all
+    OverlapBro::History.delete_all
+    OverlapBro.delete_all
+  end
+
+  describe 'records created in the same transaction' do
+    it 'are all visible in as_of() queries using the first record creation time' do
+      first_record = nil
+      second_record = nil
+      third_record = nil
+
+      Overlapper.transaction do
+        first_record = Overlapper.create!(name: 'first')
+        second_record = Overlapper.create!(name: 'second')
+        third_record = Overlapper.create!(name: 'third')
+      end
+
+      Overlapper.transaction do
+        Overlapper.create!(name: 'second transaction record')
+      end
+
+      # Get the validity start time of the first record from history
+      first_record_validity_start = first_record.history.last.validity.begin
+
+      # All records created in the same transaction should be visible
+      # when querying as_of() with the first record's creation timestamp
+      records_at_first_timestamp = Overlapper.as_of(first_record_validity_start)
+      expect(records_at_first_timestamp.map(&:id)).to contain_exactly(first_record.id, second_record.id, third_record.id)
+    end
+
+    it 'allows querying associated records created in the same transaction' do
+      overlap = nil
+
+      Overlapper.transaction do
+        bro = OverlapBro.create!(name: 'bro record')
+        overlap = Overlapper.create!(name: 'child', overlap_bro: bro)
+      end
+
+      overlap_validity_start = overlap.history.last.validity.begin
+      overlap_at_creation = Overlapper.as_of(overlap_validity_start).find(overlap.id)
+      # Child record should be visible through the association
+      expect(overlap_at_creation.overlap_bro.name).to eq('bro record')
+    end
+
+    it 'maintains consistent timestamps for records created together' do
+      records = nil
+
+      Overlapper.transaction do
+        records = Array.new(5) { |i| Overlapper.create!(name: "record-#{i}") }
+      end
+
+      # All records should have the same validity start time (transaction time)
+      validity_starts = records.map { |r| r.history.last.validity.begin }
+
+      expect(validity_starts.uniq.size).to eq(1),
+                                           "Expected all records to have the same validity start time, but got: #{validity_starts.inspect}"
+    end
+  end
+
+  describe 'records updated in the same transaction' do
+    # This test reproduces the issue where clock_timestamp() in UPDATE triggers
+    # causes records updated within the same transaction to have different timestamps,
+    # making them invisible in as_of() queries.
+    #
+    # Scenario:
+    # 1. Create an "overlap_bro" record
+    # 2. Create associated "overlappers" that reference this version
+    # 3. Update the overlappers within the same transaction
+    # 4. Query as_of(overlap_bro.created_at) should show all overlappers
+    it 'updated records are visible in as_of() queries using the transaction start time' do
+      bro = nil
+
+      OverlapBro.transaction do
+        bro = OverlapBro.create!(name: 'bro_v1')
+        3.times do |i|
+          overlap = Overlapper.create!(name: "overlapper_#{i}", overlap_bro: bro)
+          overlap.update!(name: "overlapper_#{i}_updated")
+        end
+      end
+
+      bro_created_at = bro.history.last.validity.begin
+
+      # Query all overlappers as of the bro's creation time
+      # With clock_timestamp() in UPDATE trigger, updated overlappers would have
+      # a later timestamp and would NOT be visible here
+      overlappers_at_bro_time = Overlapper.as_of(bro_created_at).where(overlap_bro_id: bro.id)
+
+      expect(overlappers_at_bro_time.count).to eq(3),
+                                               "Expected 3 overlappers visible at bro creation time, got #{overlappers_at_bro_time.count}"
+
+      expect(overlappers_at_bro_time.map(&:name)).to contain_exactly(
+        'overlapper_0_updated',
+        'overlapper_1_updated',
+        'overlapper_2_updated'
+      )
+    end
+
+    it 'multiple updates in same transaction maintain consistent timestamp' do
+      overlap = nil
+
+      Overlapper.transaction do
+        overlap = Overlapper.create!(name: 'original')
+        overlap.update!(name: 'update1')
+        overlap.update!(name: 'update2')
+        overlap.update!(name: 'final')
+      end
+
+      # All history entries created in this transaction should share the same timestamp
+      # (due to squashing or consistent now() usage)
+      history_entries = overlap.history.order(:hid)
+
+      # The final state should be visible at the creation timestamp
+      creation_time = history_entries.first.validity.begin
+      overlap_at_creation = Overlapper.as_of(creation_time).find(overlap.id)
+      expect(overlap_at_creation.name).to eq('final')
+    end
+  end
+
+  describe 'records updated after initial creation' do
+    it 'handles updates without timestamp conflicts' do
+      overlap = Overlapper.create!(name: 'original')
+
+      # Sleep a tiny bit to ensure different transaction time
+      sleep(0.01)
+
+      overlap.update!(name: 'updated')
+
+      # The update should succeed and create a valid history entry
+      overlap.reload
+      expect(overlap.name).to eq('updated')
+
+      # History should have entries with non-overlapping validity ranges
+      history_entries = overlap.history.order(:hid)
+      expect(history_entries.size).to eq 2
+    end
+  end
+end
diff --git a/spec/support/time_machine/structure.rb b/spec/support/time_machine/structure.rb
@@ -83,6 +83,16 @@ def up
       t.references :sub_bar
     end
 
+    adapter.create_table 'overlap_bros', temporal: true do |t|
+      t.string :name
+    end
+
+    adapter.create_table 'overlappers', temporal: true do |t|
+      t.string :name
+
+      t.references :overlap_bro
+    end
+
     class ::Bar < ActiveRecord::Base
       include ChronoModel::TimeMachine
 
@@ -162,6 +172,18 @@ class ::SubSubBar < ActiveRecord::Base
       belongs_to :sub_bar
     end
 
+    class ::OverlapBro < ActiveRecord::Base
+      include ChronoModel::TimeMachine
+
+      has_many :overlappers
+    end
+
+    class ::Overlapper < ActiveRecord::Base
+      include ChronoModel::TimeMachine
+
+      belongs_to :overlap_bro, optional: true
+    end
+
     # Master timeline, used in multiple specs. It is defined here
     # as a global variable to be able to be shared across specs.
     #
