History-free pre_pack: delete an object's old refs rows before inserting (fixes #482)

CHANGES.rst

@@ -5,7 +5,14 @@
 4.3.1 (unreleased)
 ==================
 
-- Nothing changed yet.
+- History-free ``pre_pack``: queue an object's deletes from
+  ``object_refs_added`` and ``object_ref`` before its new
+  ``object_refs_added`` row. The batcher runs deletes before inserts
+  only within one flush, and inserting that row can trigger a flush,
+  so the insert could reach the database while the object's old row
+  was still there, failing with ``duplicate key value violates unique
+  constraint "object_refs_added_pkey"``. Any object changed since the
+  previous pack can hit this. See :issue:`482`.
 
 
 4.3.0 (2026-05-20)

src/relstorage/adapters/packundo.py

@@ -1358,11 +1358,13 @@ def _add_refs_for_oids(self, load_batcher, store_batcher,
         # oids should be a slice of an ``OidList``, which may be an
         # ``array.array``; those are relatively slow to iterate.
 
-        # The batcher always does deletes before inserts, which is
-        # exactly what we want.
-        # In the past, we performed all deletes and then all inserts;
-        # now, things to batching, they could be interleaved, but
-        # because we process OID-by-OID, that should be fine.
+        # The batcher does deletes before inserts *within a flush*, but a
+        # flush can happen as soon as any row is queued. So for each OID we
+        # queue its deletes before its inserts: an insert queued first can
+        # fill the batch and reach the database while the OID's old
+        # ``object_refs_added`` row is still there (a duplicate key; see
+        # :issue:`482`). That row exists for every object changed since the
+        # last pack, because the table is kept between packs.
         # In the past, we also DELETED from object_refs_added and object_ref
         # everything found in the ``oids`` parameter; now we only do a delete if
         # we get back a row from object_state; again, that shouldn't matter, rows
@@ -1385,13 +1387,6 @@ def _add_refs_for_oids(self, load_batcher, store_batcher,
             state = self.driver.binary_column_as_state_type(state)
             row = (from_oid, tid)
 
-            store_batcher.insert_into(
-                'object_refs_added (zoid, tid)',
-                object_refs_added_schema,
-                row,
-                row,
-                size=2
-            )
             store_batcher.delete_from(
                 'object_refs_added',
                 zoid=from_oid
@@ -1400,6 +1395,13 @@ def _add_refs_for_oids(self, load_batcher, store_batcher,
                 'object_ref',
                 zoid=from_oid
             )
+            store_batcher.insert_into(
+                'object_refs_added (zoid, tid)',
+                object_refs_added_schema,
+                row,
+                row,
+                size=2
+            )
 
             if state:
                 try:

src/relstorage/adapters/tests/test_packundo.py

@@ -0,0 +1,93 @@
+##############################################################################
+#
+# Copyright (c) 2026 Zope Foundation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+
+from relstorage.tests import TestCase
+from relstorage.tests import MockCursor
+
+
+class _Driver(object):
+
+    @staticmethod
+    def binary_column_as_state_type(state):
+        return state
+
+
+class _LoadBatcher(object):
+
+    def __init__(self, rows):
+        self.rows = rows
+
+    def select_from(self, columns, table, suffix='', **kw):
+        return iter(self.rows)
+
+
+class HistoryFreeAddRefsForOidsTests(TestCase):
+
+    def _makeOne(self):
+        from relstorage.adapters.packundo import HistoryFreePackUndo
+        packundo = HistoryFreePackUndo.__new__(HistoryFreePackUndo)
+        packundo.driver = _Driver()
+        return packundo
+
+    def _executed_tables(self, cursor):
+        return [
+            ' '.join(stmt.split()[:3])
+            for stmt, _ in cursor.executed
+        ]
+
+    def test_old_refs_added_row_is_deleted_before_the_new_one_is_inserted(self):
+        # An object whose references were read at an older tid and that has
+        # changed since: ``object_refs_added`` still holds its old row, and
+        # the new one has the same primary key (zoid). The batcher executes
+        # deletes before inserts *within* a flush, but a flush can happen as
+        # soon as a row is queued; if the insert is queued first and fills
+        # the batch, it reaches the database while the old row is still
+        # there: "duplicate key value violates unique constraint
+        # object_refs_added_pkey" (#482). row_limit=1 flushes on every row.
+        from relstorage.adapters.batch import RowBatcher
+        cursor = MockCursor()
+        store_batcher = RowBatcher(cursor, row_limit=1)
+        packundo = self._makeOne()
+
+        packundo._add_refs_for_oids(
+            _LoadBatcher([(1, 5, b'state')]),
+            store_batcher,
+            [1],
+            lambda state: [2],
+        )
+        store_batcher.flush()
+
+        executed = self._executed_tables(cursor)
+        delete = executed.index('DELETE FROM object_refs_added')
+        insert = executed.index('INSERT INTO object_refs_added')
+        self.assertLess(delete, insert, executed)
+
+    def test_old_object_ref_rows_are_deleted_before_the_new_ones_are_inserted(self):
+        from relstorage.adapters.batch import RowBatcher
+        cursor = MockCursor()
+        store_batcher = RowBatcher(cursor, row_limit=1)
+        packundo = self._makeOne()
+
+        packundo._add_refs_for_oids(
+            _LoadBatcher([(1, 5, b'state')]),
+            store_batcher,
+            [1],
+            lambda state: [2],
+        )
+        store_batcher.flush()
+
+        executed = self._executed_tables(cursor)
+        delete = executed.index('DELETE FROM object_ref')
+        insert = executed.index('INSERT INTO object_ref')
+        self.assertLess(delete, insert, executed)