Ensure strong ref to obj before calling persistent_to_deleted, others

Add checks in spots where state.obj() might be late-GC'ed before
we get a chance to call the event.   There may be more cases
of these which we should address as they come up.  The Session
should always be maintaining strong refs to objects that have
pending operations left on them, so for these cases we need
to ensure that ref remains long enough for the event to be called.

Change-Id: I1a7c7bc57130acc11f54ad55924af2e36ac75101
Fixes: #3808

4 files changed, 95 insertions, 6 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index c2dd2d848..a3cc96f99 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,14 @@
     :version: 1.1.0
 
     .. change::
+        :tags: bug, orm
+        :tickets: 3808
+
+        Fixed bug in new :meth:`.SessionEvents.persistent_to_deleted` event
+        where the target object could be garbage collected before the event
+        is fired off.
+
+    .. change::
         :tags: bug, sql
         :tickets: 3809
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 4a65e0719..b492cbb7f 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1642,6 +1642,11 @@ class Session(_SessionClassMethods):
             if self._enable_transaction_accounting and self.transaction:
                 self.transaction._deleted[state] = True
 
+            if persistent_to_deleted is not None:
+                # get a strong reference before we pop out of
+                # self._deleted
+                obj = state.obj()
+
             self.identity_map.safe_discard(state)
             self._deleted.pop(state, None)
             state._deleted = True
@@ -1649,7 +1654,7 @@ class Session(_SessionClassMethods):
             # is still in the transaction snapshot and needs to be
             # tracked as part of that
             if persistent_to_deleted is not None:
-                persistent_to_deleted(self, state.obj())
+                persistent_to_deleted(self, obj)
 
     def add(self, instance, _warn=True):
         """Place an object in the ``Session``.
@@ -1964,6 +1969,11 @@ class Session(_SessionClassMethods):
                 )
 
         obj = state.obj()
+
+        # check for late gc
+        if obj is None:
+            return
+
         to_attach = self._before_attach(state, obj)
 
         self._deleted.pop(state, None)
diff --git a/lib/sqlalchemy/orm/state.py b/lib/sqlalchemy/orm/state.py
index 2704367f9..519220e34 100644
--- a/lib/sqlalchemy/orm/state.py
+++ b/lib/sqlalchemy/orm/state.py
@@ -328,13 +328,21 @@ class InstanceState(interfaces.InspectionAttr):
             if persistent:
                 if to_transient:
                     if persistent_to_transient is not None:
-                        persistent_to_transient(session, state.obj())
+                        obj = state.obj()
+                        if obj is not None:
+                            persistent_to_transient(session, obj)
                 elif persistent_to_detached is not None:
-                    persistent_to_detached(session, state.obj())
+                    obj = state.obj()
+                    if obj is not None:
+                        persistent_to_detached(session, obj)
             elif deleted and deleted_to_detached is not None:
-                deleted_to_detached(session, state.obj())
+                obj = state.obj()
+                if obj is not None:
+                    deleted_to_detached(session, obj)
             elif pending and pending_to_transient is not None:
-                pending_to_transient(session, state.obj())
+                obj = state.obj()
+                if obj is not None:
+                    pending_to_transient(session, obj)
 
             state._strong_obj = None
 
diff --git a/test/orm/test_events.py b/test/orm/test_events.py
index ab61077ae..594aabdab 100644
--- a/test/orm/test_events.py
+++ b/test/orm/test_events.py
@@ -9,7 +9,7 @@ from sqlalchemy.orm import mapper, relationship, \
     Session, sessionmaker, attributes, configure_mappers
 from sqlalchemy.orm.instrumentation import ClassManager
 from sqlalchemy.orm import instrumentation, events
-from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_, is_not_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing.util import gc_collect
@@ -1764,6 +1764,69 @@ class SessionLifecycleEventsTest(_RemoveListeners, _fixtures.FixtureTest):
             ]
         )
 
+    def test_pending_to_persistent_del(self):
+        sess, User, start_events = self._fixture()
+
+        @event.listens_for(sess, "pending_to_persistent")
+        def pending_to_persistent(session, instance):
+            listener.flag_checked(instance)
+            # this is actually u1, because
+            # we have a strong ref internally
+            is_not_(None, instance)
+
+        u1 = User(name='u1')
+        sess.add(u1)
+
+        u1_inst_state = u1._sa_instance_state
+        del u1
+
+        gc_collect()
+
+        listener = start_events()
+
+        sess.flush()
+
+        eq_(
+            listener.mock_calls,
+            [
+                call.flag_checked(u1_inst_state.obj()),
+                call.pending_to_persistent(
+                    sess, u1_inst_state.obj()),
+            ]
+        )
+
+    def test_persistent_to_deleted_del(self):
+        sess, User, start_events = self._fixture()
+
+        u1 = User(name='u1')
+        sess.add(u1)
+        sess.flush()
+
+        listener = start_events()
+
+        @event.listens_for(sess, "persistent_to_deleted")
+        def persistent_to_deleted(session, instance):
+            is_not_(None, instance)
+            listener.flag_checked(instance)
+
+        sess.delete(u1)
+        u1_inst_state = u1._sa_instance_state
+
+        del u1
+        gc_collect()
+
+        sess.flush()
+
+        eq_(
+            listener.mock_calls,
+            [
+                call.persistent_to_deleted(sess, u1_inst_state.obj()),
+                call.flag_checked(u1_inst_state.obj())
+             ]
+         )
+
+
+
     def test_detached_to_persistent(self):
         sess, User, start_events = self._fixture()