Merge branch 'fix-443'

4 files changed, 89 insertions, 27 deletions

diff --git a/NEWS b/NEWS
index 88689f6..e28202c 100644
--- a/NEWS
+++ b/NEWS
@@ -1,6 +1,14 @@
 Current release
 ---------------
 
+What's new in psycopg 2.7.2
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- Fixed inconsistent state in externally closed connections
+  (:tickets:`#263, #311, #443`). Was fixed in 2.6.2 but not included in
+  2.7 by mistake.
+
+
 What's new in psycopg 2.7.1
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
@@ -82,6 +90,8 @@ What's new in psycopg 2.6.3
 What's new in psycopg 2.6.2
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+- Fixed inconsistent state in externally closed connections
+  (:tickets:`#263, #311, #443`).
 - Report the server response status on errors (such as :ticket:`#281`).
 - Raise `!NotSupportedError` on unhandled server response status
   (:ticket:`#352`).
@@ -91,7 +101,8 @@ What's new in psycopg 2.6.2
   (:ticket:`#333`).
 - Fixed `!PersistentConnectionPool` on Python 3 (:ticket:`#348`).
 - Fixed segfault on `repr()` of an unitialized connection (:ticket:`#361`).
-- Allow adapting bytes using QuotedString on Python 3 too (:ticket:`#365`).
+- Allow adapting bytes using `~psycopg2.extensions.QuotedString` on Python 3
+  (:ticket:`#365`).
 - Added support for setuptools/wheel (:ticket:`#370`).
 - Fix build on Windows with Python 3.5, VS 2015 (:ticket:`#380`).
 - Fixed `!errorcodes.lookup` initialization thread-safety (:ticket:`#382`).
diff --git a/psycopg/pqpath.c b/psycopg/pqpath.c
index 59cbb5e..f270ce8 100644
--- a/psycopg/pqpath.c
+++ b/psycopg/pqpath.c
@@ -179,8 +179,10 @@ pq_raise(connectionObject *conn, cursorObject *curs, PGresult **pgres)
 
     /* if the connection has somehow been broken, we mark the connection
        object as closed but requiring cleanup */
-    if (conn->pgconn != NULL && PQstatus(conn->pgconn) == CONNECTION_BAD)
+    if (conn->pgconn != NULL && PQstatus(conn->pgconn) == CONNECTION_BAD) {
         conn->closed = 2;
+        exc = OperationalError;
+    }
 
     if (pgres == NULL && curs != NULL)
         pgres = &curs->pgres;
@@ -214,9 +216,9 @@ pq_raise(connectionObject *conn, cursorObject *curs, PGresult **pgres)
     if (code != NULL) {
         exc = exception_from_sqlstate(code);
     }
-    else {
-        /* Fallback if there is no exception code (reported happening e.g.
-         * when the connection is closed). */
+    else if (exc == NULL) {
+        /* Fallback if there is no exception code (unless we already
+           determined that the connection was closed). */
         exc = DatabaseError;
     }
 
@@ -986,6 +988,9 @@ pq_execute(cursorObject *curs, const char *query, int async, int no_result, int
 
         /* don't let pgres = NULL go to pq_fetch() */
         if (curs->pgres == NULL) {
+            if (CONNECTION_BAD == PQstatus(curs->conn->pgconn)) {
+                curs->conn->closed = 2;
+            }
             pthread_mutex_unlock(&(curs->conn->lock));
             Py_BLOCK_THREADS;
             if (!PyErr_Occurred()) {
@@ -1013,6 +1018,9 @@ pq_execute(cursorObject *curs, const char *query, int async, int no_result, int
 
         CLEARPGRES(curs->pgres);
         if (PQsendQuery(curs->conn->pgconn, query) == 0) {
+            if (CONNECTION_BAD == PQstatus(curs->conn->pgconn)) {
+                curs->conn->closed = 2;
+            }
             pthread_mutex_unlock(&(curs->conn->lock));
             Py_BLOCK_THREADS;
             PyErr_SetString(OperationalError,
diff --git a/tests/test_connection.py b/tests/test_connection.py
index dae0560..385d979 100755
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -72,16 +72,8 @@ class ConnectionTests(ConnectingTestCase):
         # ticket #148
         conn = self.conn
         cur = conn.cursor()
-        try:
-            cur.execute("select pg_terminate_backend(pg_backend_pid())")
-        except psycopg2.OperationalError, e:
-            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
-                raise
-        except psycopg2.DatabaseError, e:
-            # curiously when disconnected in green mode we get a DatabaseError
-            # without pgcode.
-            if e.pgcode is not None:
-                raise
+        self.assertRaises(psycopg2.OperationalError,
+            cur.execute, "select pg_terminate_backend(pg_backend_pid())")
 
         self.assertEqual(conn.closed, 2)
         conn.close()
diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index 42e12c4..10b8d71 100755
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -27,7 +27,10 @@ import pickle
 import psycopg2
 import psycopg2.extensions
 from testutils import (unittest, ConnectingTestCase, skip_before_postgres,
-    skip_if_no_namedtuple, skip_if_no_getrefcount, slow)
+    skip_if_no_namedtuple, skip_if_no_getrefcount, slow, skip_if_no_superuser,
+    skip_if_windows)
+
+import psycopg2.extras
 
 
 class CursorTests(ConnectingTestCase):
@@ -518,28 +521,76 @@ class CursorTests(ConnectingTestCase):
             RETURNS INT AS
                 'SELECT $1 * $1'
             LANGUAGE SQL
-        ''' % (procname, escaped_paramname));
+        ''' % (procname, escaped_paramname))
 
         # Make sure callproc works right
-        cur.callproc(procname, { paramname: 2 })
+        cur.callproc(procname, {paramname: 2})
         self.assertEquals(cur.fetchone()[0], 4)
 
         # Make sure callproc fails right
         failing_cases = [
-            ({ paramname: 2, 'foo': 'bar' }, psycopg2.ProgrammingError),
-            ({ paramname: '2' },             psycopg2.ProgrammingError),
-            ({ paramname: 'two' },           psycopg2.ProgrammingError),
-            ({ u'bj\xc3rn': 2 },             psycopg2.ProgrammingError),
-            ({ 3: 2 },                       TypeError),
-            ({ self: 2 },                    TypeError),
+            ({paramname: 2, 'foo': 'bar'}, psycopg2.ProgrammingError),
+            ({paramname: '2'}, psycopg2.ProgrammingError),
+            ({paramname: 'two'}, psycopg2.ProgrammingError),
+            ({u'bj\xc3rn': 2}, psycopg2.ProgrammingError),
+            ({3: 2}, TypeError),
+            ({self: 2}, TypeError),
         ]
         for parameter_sequence, exception in failing_cases:
             self.assertRaises(exception, cur.callproc, procname, parameter_sequence)
             self.conn.rollback()
 
-    def test_callproc_badparam(self):
-        cur = self.conn.cursor()
-        self.assertRaises(TypeError, cur.callproc, 'lower', 42)
+    @skip_if_no_superuser
+    @skip_if_windows
+    @skip_before_postgres(8, 4)
+    def test_external_close_sync(self):
+        # If a "victim" connection is closed by a "control" connection
+        # behind psycopg2's back, psycopg2 always handles it correctly:
+        # raise OperationalError, set conn.closed to 2. This reproduces
+        # issue #443, a race between control_conn closing victim_conn and
+        # psycopg2 noticing.
+        control_conn = self.conn
+        connect_func = self.connect
+        wait_func = lambda conn: None
+        self._test_external_close(control_conn, connect_func, wait_func)
+
+    @skip_if_no_superuser
+    @skip_if_windows
+    @skip_before_postgres(8, 4)
+    def test_external_close_async(self):
+        # Issue #443 is in the async code too. Since the fix is duplicated,
+        # so is the test.
+        control_conn = self.conn
+        connect_func = lambda: self.connect(async=True)
+        wait_func = psycopg2.extras.wait_select
+        self._test_external_close(control_conn, connect_func, wait_func)
+
+    def _test_external_close(self, control_conn, connect_func, wait_func):
+        # The short sleep before using victim_conn the second time makes it
+        # much more likely to lose the race and see the bug. Repeating the
+        # test several times makes it even more likely.
+        for i in range(10):
+            victim_conn = connect_func()
+            wait_func(victim_conn)
+
+            with victim_conn.cursor() as cur:
+                cur.execute('select pg_backend_pid()')
+                wait_func(victim_conn)
+                pid1 = cur.fetchall()[0][0]
+
+            with control_conn.cursor() as cur:
+                cur.execute('select pg_terminate_backend(%s)', (pid1,))
+
+            time.sleep(0.001)
+
+            def f():
+                with victim_conn.cursor() as cur:
+                    cur.execute('select 1')
+                    wait_func(victim_conn)
+
+            self.assertRaises(psycopg2.OperationalError, f)
+
+            self.assertEqual(victim_conn.closed, 2)
 
 
 def test_suite():