[ruby] expose driver timestamp semantics

ruby/driver/riverqueue-activerecord/lib/driver.rb

@@ -164,7 +164,7 @@ def transaction(&)
     private_constant :SQLITE_CONFLICT_WHERE, :SQLITE_JOB_COLUMNS, :UNIQUE_INSERT_METADATA_KEY
 
     private def format_time(time)
-      time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N")
+      normalize_time(time).strftime("%Y-%m-%d %H:%M:%S.%3N")
     end
 
     private def notification_query(sql)

ruby/driver/riverqueue-sequel/lib/driver.rb

@@ -155,7 +155,7 @@ def transaction
     private_constant :SQLITE_CONFLICT_WHERE, :SQLITE_JOB_COLUMNS, :UNIQUE_INSERT_METADATA_KEY
 
     private def format_time(time)
-      time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N")
+      normalize_time(time).strftime("%Y-%m-%d %H:%M:%S.%3N")
     end
 
     private def postgres_insert_params_to_hash(insert_params, nonce)

ruby/lib/driver/runtime.rb

@@ -5,12 +5,19 @@
 module River::Driver
   # Database operations used by River's worker runtime. Driver gems implement a
   # small set of raw SQL primitives and include this module so the state machine
-  # stays identical across ActiveRecord and Sequel.
+  # stays identical across ActiveRecord and Sequel. Those private primitives
+  # provide execution, quoting, row access, and connection management inside
+  # driver modules; callers use the driver's public operations and Ruby values.
   module Runtime
     def init_driver
       postgres_capabilities if runtime_postgres?
     end
 
+    # Returns the UTC timestamp that writing and reading +time+ preserves.
+    def normalize_time(time)
+      runtime_postgres? ? time.getutc.floor(6) : time.getutc.round(3)
+    end
+
     def postgres_capabilities
       @postgres_capabilities_cache.fetch(runtime_connection_pool) do
         row = runtime_query_rows(<<~SQL).first
@@ -471,6 +478,11 @@ def request_resign
       runtime_query_rows(sql)
     end
 
+    # Smallest timestamp increment preserved by this driver, in seconds.
+    def time_precision
+      runtime_postgres? ? Rational(1, 1_000_000) : Rational(1, 1_000)
+    end
+
     private def runtime_append_error(error)
       value = error.respond_to?(:to_h) ? error.to_h : error
       if runtime_postgres?
@@ -782,9 +794,9 @@ def request_resign
 
       cast = runtime_postgres? ? "::timestamptz" : ""
       encoded = if runtime_postgres?
-        value.getutc.iso8601(6)
+        normalize_time(value).iso8601(6)
       else
-        value.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N")
+        normalize_time(value).strftime("%Y-%m-%d %H:%M:%S.%3N")
       end
       "#{runtime_quote(encoded)}#{cast}"
     end

ruby/sig/driver.rbs

@@ -28,6 +28,7 @@ module River
     def leader_acquire: (String, ?ttl: Integer, ?now: Time) -> bool
     def leader_release: (String) -> untyped
     def leader_renew: (String, ?ttl: Integer, ?now: Time) -> bool
+    def normalize_time: (Time) -> Time
     def notification_delete_before: (horizon: Time, ?max: Integer) -> Integer
     def notification_listener: () -> untyped
     def request_resign: () -> nil
@@ -37,6 +38,7 @@ module River
     def queue_resume: (String, ?now: Time) -> Array[Queue]
     def queue_update: (String, metadata: Hash[untyped, untyped], ?now: Time) -> Queue?
     def queue_upsert: (String, ?metadata: Hash[untyped, untyped], ?now: Time) -> Queue?
+    def time_precision: () -> Rational
     def transaction: [T] () { () -> T } -> T
 
     # this set of methods is used only in tests
@@ -126,6 +128,7 @@ module River
       def runtime_decode_postgres_job: (untyped) -> JobRow
       @postgres_capabilities_cache: PostgresCapabilities::Cache
       def init_driver: () -> void
+      def normalize_time: (Time) -> Time
       def postgres_capabilities: () -> PostgresCapabilities
       def postgres_notify_insert: (Array[JobInsertParams]) -> void
       def job_get_cancelled_ids: (Array[Integer]) -> Array[Integer]
@@ -158,6 +161,8 @@ module River
       def queue_update: (untyped, metadata: untyped, ?now: untyped) -> untyped
       def queue_upsert: (untyped, ?metadata: untyped, ?now: untyped) -> untyped
 
+      def time_precision: () -> Rational
+
       private
 
       def job_get_by_id: (untyped) -> untyped

ruby/spec/driver_runtime_feature_spec.rb

@@ -13,6 +13,14 @@
   it_behaves_like "driver queue and leadership state"
 end
 
+RSpec.describe "SQLite driver timestamp contract", database: :sqlite do
+  around(:each) { |example| sqlite_test_transaction(&example) }
+
+  let(:driver) { River::Driver::Sequel.new(SQLITE_DB) }
+
+  it_behaves_like "driver timestamp precision"
+end
+
 RSpec.describe "SQLite numeric metadata counters", database: :sqlite do
   around(:each) { |example| sqlite_test_transaction(&example) }
 

ruby/spec/driver_runtime_shared_examples.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require "delegate"
+
 class DriverRuntimeArgs
   def initialize(value)
     @value = value
@@ -10,6 +12,55 @@ def kind = "driver_runtime"
   def to_json = JSON.dump(value: @value)
 end
 
+RSpec.shared_examples "driver timestamp precision" do
+  let(:client) { River::Client.new(driver) }
+
+  describe "timestamp precision" do
+    it "exposes timestamp operations through a delegating driver" do
+      delegated = SimpleDelegator.new(driver)
+      time = Time.utc(2026, 10, 25) + Rational(123_500_001, 1_000_000_000)
+
+      expect(delegated.normalize_time(time)).to eq(driver.normalize_time(time))
+      expect(delegated.time_precision).to eq(driver.time_precision)
+    end
+
+    [
+      ["a timestamp just below the millisecond rounding boundary", 123_499_999, Rational(123_499, 1_000_000), Rational(123, 1_000)],
+      ["a timestamp just above the millisecond rounding boundary", 123_500_001, Rational(123_500, 1_000_000), Rational(124, 1_000)],
+      ["a timestamp just before a second rollover", 999_999_999, Rational(999_999, 1_000_000), 1]
+    ].each do |name, nanoseconds, postgres_fraction, sqlite_fraction|
+      it "normalizes #{name} to its persisted UTC value" do
+        midnight = Time.utc(2026, 10, 25)
+        expected_fraction = (driver.migration_backend == :sqlite) ? sqlite_fraction : postgres_fraction
+        time = (midnight + Rational(nanoseconds, 1_000_000_000)).getlocal("+05:30")
+        expected = midnight + expected_fraction
+        normalized = driver.normalize_time(time)
+        driver.queue_upsert("precision", now: time)
+
+        expect(normalized).to eq(expected)
+        expect(normalized).to be_utc
+        expect(driver.normalize_time(normalized)).to eq(normalized)
+        expect(driver.queue_get("precision").created_at).to eq(expected)
+        inserted = client.insert(DriverRuntimeArgs.new(1), scheduled_at: time).job
+        expect(client.job_get(inserted.id).scheduled_at).to eq(expected)
+      end
+    end
+
+    it "preserves one timestamp increment when written and read back" do
+      expected = (driver.migration_backend == :sqlite) ? Rational(1, 1_000) : Rational(1, 1_000_000)
+      expect(driver.time_precision).to be_a(Rational)
+      expect(driver.time_precision).to eq(expected)
+      time = driver.normalize_time(Time.utc(2026, 10, 25) + Rational(123_500_001, 1_000_000_000))
+      driver.queue_upsert("precision", now: time)
+      driver.queue_update("precision", metadata: {}, now: time + driver.time_precision)
+
+      queue = driver.queue_get("precision")
+      expect(queue.updated_at.to_r - queue.created_at.to_r).to eq(driver.time_precision)
+      expect(queue.updated_at).to be > queue.created_at
+    end
+  end
+end
+
 RSpec.shared_examples "overflowing metadata counters" do
   let(:client) { River::Client.new(driver) }
   let(:now) { Time.utc(2026, 1, 2, 3, 4, 5) }
@@ -43,6 +94,7 @@ def to_json = JSON.dump(value: @value)
 end
 
 RSpec.shared_examples "driver job state machine" do
+  it_behaves_like "driver timestamp precision"
   it_behaves_like "overflowing metadata counters"
 
   let(:client) { River::Client.new(driver) }