diff --git a/asyncpg/_testbase/__init__.py b/asyncpg/_testbase/__init__.py
index 95775e11..96d43403 100644
--- a/asyncpg/_testbase/__init__.py
+++ b/asyncpg/_testbase/__init__.py
@@ -267,6 +267,7 @@ def _shutdown_cluster(cluster):
 
 
 def create_pool(dsn=None, *,
+                init_size=10,
                 min_size=10,
                 max_size=10,
                 max_queries=50000,
@@ -281,6 +282,7 @@ def create_pool(dsn=None, *,
                 **connect_kwargs):
     return pool_class(
         dsn,
+        init_size=init_size,
         min_size=min_size,
         max_size=max_size,
         max_queries=max_queries,
diff --git a/asyncpg/pool.py b/asyncpg/pool.py
index 5c7ea9ca..a118cced 100644
--- a/asyncpg/pool.py
+++ b/asyncpg/pool.py
@@ -293,6 +293,15 @@ def _deactivate_inactive_connection(self) -> None:
                 'attempting to deactivate an acquired connection')
 
         if self._con is not None:
+            # The connection is idle and not in use,
+            # but we have min size limitation. So keep it alive for a while.
+            if self._pool.get_size() <= self._pool.get_min_size():
+                # We already in the callback. Clean the field
+                self._inactive_callback = None
+                # But next time it can be the case when we have to terminate it
+                self._setup_inactive_callback()
+                return
+
             # The connection is idle and not in use, so it's fine to
             # use terminate() instead of close().
             self._con.terminate()
@@ -338,7 +347,7 @@ class Pool:
     """
 
     __slots__ = (
-        '_queue', '_loop', '_minsize', '_maxsize',
+        '_queue', '_loop', '_initsize', '_minsize', '_maxsize',
         '_init', '_connect', '_reset', '_connect_args', '_connect_kwargs',
         '_holders', '_initialized', '_initializing', '_closing',
         '_closed', '_connection_class', '_record_class', '_generation',
@@ -346,6 +355,7 @@ class Pool:
     )
 
     def __init__(self, *connect_args,
+                 init_size,
                  min_size,
                  max_size,
                  max_queries,
@@ -381,6 +391,16 @@ def __init__(self, *connect_args,
         if min_size > max_size:
             raise ValueError('min_size is greater than max_size')
 
+        if init_size < 0:
+            raise ValueError(
+                'init_size is expected to be greater or equal to zero')
+
+        if init_size > max_size:
+            raise ValueError('init_size is greater than max_size')
+
+        if init_size < min_size:
+            raise ValueError('init_size is smaller than min_size')
+
         if max_queries <= 0:
             raise ValueError('max_queries is expected to be greater than zero')
 
@@ -399,6 +419,7 @@ def __init__(self, *connect_args,
                 'record_class is expected to be a subclass of '
                 'asyncpg.Record, got {!r}'.format(record_class))
 
+        self._initsize = init_size
         self._minsize = min_size
         self._maxsize = max_size
 
@@ -454,7 +475,7 @@ async def _initialize(self):
             self._holders.append(ch)
             self._queue.put_nowait(ch)
 
-        if self._minsize:
+        if self._initsize:
             # Since we use a LIFO queue, the first items in the queue will be
             # the last ones in `self._holders`.  We want to pre-connect the
             # first few connections in the queue, therefore we want to walk
@@ -465,11 +486,11 @@ async def _initialize(self):
             first_ch = self._holders[-1]  # type: PoolConnectionHolder
             await first_ch.connect()
 
-            if self._minsize > 1:
+            if self._initsize > 1:
                 connect_tasks = []
                 for i, ch in enumerate(reversed(self._holders[:-1])):
-                    # `minsize - 1` because we already have first_ch
-                    if i >= self._minsize - 1:
+                    # `initsize - 1` because we already have first_ch
+                    if i >= self._initsize - 1:
                         break
                     connect_tasks.append(ch.connect())
 
@@ -489,10 +510,21 @@ def get_size(self):
         """
         return sum(h.is_connected() for h in self._holders)
 
+    def get_init_size(self):
+        """Return the initial number of connections in this pool.
+
+        .. versionadded:: 0.32.0
+        """
+        return self._initsize
+
     def get_min_size(self):
         """Return the minimum number of connections in this pool.
 
         .. versionadded:: 0.25.0
+
+        .. versionchanged:: 0.32.0
+            The parameter now controls the connection floor rather than the
+            initial pool size (see ``init_size``).
         """
         return self._minsize
 
@@ -1073,6 +1105,7 @@ def __await__(self):
 
 
 def create_pool(dsn=None, *,
+                init_size=10,
                 min_size=10,
                 max_size=10,
                 max_queries=50000,
@@ -1147,8 +1180,11 @@ def create_pool(dsn=None, *,
         the connections in this pool.  Must be a subclass of
         :class:`~asyncpg.Record`.
 
+    :param int init_size:
+        Number of connections the pool will be initialized with.
+
     :param int min_size:
-        Number of connection the pool will be initialized with.
+        Minimum number of connections the pool will keep alive at all times.
 
     :param int max_size:
         Max number of connections in the pool.
@@ -1230,11 +1266,18 @@ def create_pool(dsn=None, *,
 
     .. versionchanged:: 0.30.0
        Added the *connect* and *reset* parameters.
+
+    .. versionchanged:: 0.32.0
+       The *min_size* parameter now defines the connection floor (minimum
+       number of live connections kept at all times).  The former role of
+       *min_size* — setting the initial pool size — is now handled by the
+       new *init_size* parameter.
     """
     return Pool(
         dsn,
         connection_class=connection_class,
         record_class=record_class,
+        init_size=init_size,
         min_size=min_size,
         max_size=max_size,
         max_queries=max_queries,
diff --git a/pyproject.toml b/pyproject.toml
index e2b18388..e6e285cb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -62,7 +62,8 @@ docs = [
 [build-system]
 requires = [
     "setuptools>=77.0.3",
-    "Cython(>=3.2.1,<4.0.0)"
+    "Cython(>=3.2.1,<4.0.0)",
+    "packaging",
 ]
 build-backend = "setuptools.build_meta"
 
diff --git a/setup.py b/setup.py
index f9fafadf..247fe596 100644
--- a/setup.py
+++ b/setup.py
@@ -188,8 +188,6 @@ def finalize_options(self):
                         need_cythonize = True
 
         if need_cythonize:
-            import pkg_resources
-
             # Double check Cython presence in case setup_requires
             # didn't go into effect (most likely because someone
             # imported Cython before setup_requires injected the
@@ -201,8 +199,9 @@ def finalize_options(self):
                     'please install {} to compile asyncpg from source'.format(
                         CYTHON_DEPENDENCY))
 
-            cython_dep = pkg_resources.Requirement.parse(CYTHON_DEPENDENCY)
-            if Cython.__version__ not in cython_dep:
+            from packaging.requirements import Requirement
+            cython_dep = Requirement(CYTHON_DEPENDENCY)
+            if Cython.__version__ not in cython_dep.specifier:
                 raise RuntimeError(
                     'asyncpg requires {}, got Cython=={}'.format(
                         CYTHON_DEPENDENCY, Cython.__version__
diff --git a/tests/test_adversity.py b/tests/test_adversity.py
index a6e03feb..6263be8a 100644
--- a/tests/test_adversity.py
+++ b/tests/test_adversity.py
@@ -29,7 +29,7 @@ async def test_connection_close_timeout(self):
     @tb.with_timeout(30.0)
     async def test_pool_acquire_timeout(self):
         pool = await self.create_pool(
-            database='postgres', min_size=2, max_size=2)
+            database='postgres', init_size=2, min_size=0, max_size=2)
         try:
             self.proxy.trigger_connectivity_loss()
             for _ in range(2):
@@ -46,7 +46,7 @@ async def test_pool_acquire_timeout(self):
     @tb.with_timeout(30.0)
     async def test_pool_release_timeout(self):
         pool = await self.create_pool(
-            database='postgres', min_size=2, max_size=2)
+            database='postgres', init_size=2, min_size=0, max_size=2)
         try:
             with self.assertRaises(asyncio.TimeoutError):
                 async with pool.acquire(timeout=0.5):
@@ -74,7 +74,8 @@ def kill_connectivity():
             self.proxy.trigger_connectivity_loss()
 
         new_pool = self.create_pool(
-            database='postgres', min_size=pool_size, max_size=pool_size,
+            database='postgres',
+            init_size=pool_size, min_size=0, max_size=pool_size,
             timeout=cmd_timeout, command_timeout=cmd_timeout)
 
         with self.assertRunUnder(worst_runtime):
diff --git a/tests/test_cache_invalidation.py b/tests/test_cache_invalidation.py
index 5cab2d92..f2901c9e 100644
--- a/tests/test_cache_invalidation.py
+++ b/tests/test_cache_invalidation.py
@@ -77,7 +77,7 @@ async def test_prepare_cache_invalidation_in_transaction(self):
 
     async def test_prepare_cache_invalidation_in_pool(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=2, max_size=2)
+                                      init_size=2, min_size=0, max_size=2)
 
         await self.con.execute('CREATE TABLE tab1(a int, b int)')
 
@@ -309,10 +309,10 @@ async def test_type_cache_invalidation_on_change_attr(self):
     async def test_type_cache_invalidation_in_pool(self):
         await self.con.execute('CREATE DATABASE testdb')
         pool = await self.create_pool(database='postgres',
-                                      min_size=2, max_size=2)
+                                      init_size=2, min_size=0, max_size=2)
 
         pool_chk = await self.create_pool(database='testdb',
-                                          min_size=2, max_size=2)
+                                          init_size=2, min_size=0, max_size=2)
 
         await self.con.execute('CREATE TYPE typ1 AS (x int, y int)')
         await self.con.execute('CREATE TABLE tab1(a int, b typ1)')
diff --git a/tests/test_connect.py b/tests/test_connect.py
index 955fb825..247f4696 100644
--- a/tests/test_connect.py
+++ b/tests/test_connect.py
@@ -1961,6 +1961,7 @@ async def test_ssl_connection_pool(self):
             host='localhost',
             user='ssl_user',
             database='postgres',
+            init_size=5,
             min_size=5,
             max_size=10,
             ssl=ssl_context)
@@ -2221,6 +2222,7 @@ async def test_nossl_connection_pool(self):
             host='localhost',
             user='ssl_user',
             database='postgres',
+            init_size=5,
             min_size=5,
             max_size=10,
             ssl='prefer')
diff --git a/tests/test_pool.py b/tests/test_pool.py
index 695363b7..27e90123 100644
--- a/tests/test_pool.py
+++ b/tests/test_pool.py
@@ -47,7 +47,9 @@ async def test_pool_01(self):
         for n in {1, 5, 10, 20, 100}:
             with self.subTest(tasksnum=n):
                 pool = await self.create_pool(database='postgres',
-                                              min_size=5, max_size=10)
+                                              init_size=5,
+                                              min_size=1,
+                                              max_size=10)
 
                 async def worker():
                     con = await pool.acquire()
@@ -62,7 +64,9 @@ async def test_pool_02(self):
         for n in {1, 3, 5, 10, 20, 100}:
             with self.subTest(tasksnum=n):
                 async with self.create_pool(database='postgres',
-                                            min_size=5, max_size=5) as pool:
+                                            init_size=5,
+                                            min_size=1,
+                                            max_size=5) as pool:
 
                     async def worker():
                         con = await pool.acquire(timeout=5)
@@ -74,7 +78,7 @@ async def worker():
 
     async def test_pool_03(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         con = await pool.acquire(timeout=1)
         with self.assertRaises(asyncio.TimeoutError):
@@ -85,7 +89,7 @@ async def test_pool_03(self):
 
     async def test_pool_04(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         con = await pool.acquire(timeout=POOL_NOMINAL_TIMEOUT)
 
@@ -110,7 +114,9 @@ async def test_pool_05(self):
         for n in {1, 3, 5, 10, 20, 100}:
             with self.subTest(tasksnum=n):
                 pool = await self.create_pool(database='postgres',
-                                              min_size=5, max_size=10)
+                                              init_size=5,
+                                              min_size=1,
+                                              max_size=10)
 
                 async def worker():
                     async with pool.acquire() as con:
@@ -127,7 +133,7 @@ async def setup(con):
             fut.set_result(con)
 
         async with self.create_pool(database='postgres',
-                                    min_size=5, max_size=5,
+                                    init_size=5, min_size=1, max_size=5,
                                     setup=setup) as pool:
             async with pool.acquire() as con:
                 pass
@@ -169,8 +175,8 @@ async def user(pool):
                     raise RuntimeError('init was not called')
 
         async with self.create_pool(database='postgres',
-                                    min_size=2,
-                                    max_size=5,
+                                    init_size=2,
+                                    min_size=1, max_size=5,
                                     connect=connect,
                                     init=init,
                                     setup=setup,
@@ -196,7 +202,7 @@ async def bad_connect(*args, **kwargs):
 
     async def test_pool_08(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         con = await pool.acquire(timeout=POOL_NOMINAL_TIMEOUT)
         with self.assertRaisesRegex(asyncpg.InterfaceError, 'is not a member'):
@@ -204,10 +210,10 @@ async def test_pool_08(self):
 
     async def test_pool_09(self):
         pool1 = await self.create_pool(database='postgres',
-                                       min_size=1, max_size=1)
+                                       init_size=1, min_size=1, max_size=1)
 
         pool2 = await self.create_pool(database='postgres',
-                                       min_size=1, max_size=1)
+                                       init_size=1, min_size=1, max_size=1)
 
         try:
             con = await pool1.acquire(timeout=POOL_NOMINAL_TIMEOUT)
@@ -222,7 +228,7 @@ async def test_pool_09(self):
 
     async def test_pool_10(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         con = await pool.acquire()
         await pool.release(con)
@@ -232,7 +238,7 @@ async def test_pool_10(self):
 
     async def test_pool_11(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         async with pool.acquire() as con:
             self.assertIn(repr(con._con), repr(con))  # Test __repr__.
@@ -289,7 +295,7 @@ async def test_pool_11(self):
 
     async def test_pool_12(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         async with pool.acquire() as con:
             self.assertTrue(isinstance(con, pg_connection.Connection))
@@ -299,7 +305,7 @@ async def test_pool_12(self):
 
     async def test_pool_13(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         async with pool.acquire() as con:
             self.assertIn('Execute an SQL command', con.execute.__doc__)
@@ -335,7 +341,7 @@ async def setup(con):
             last_con = None
             cons = []
             async with self.create_pool(database='postgres',
-                                        min_size=1, max_size=1,
+                                        init_size=1, min_size=1, max_size=1,
                                         setup=setup) as pool:
                 with self.assertRaises(Error):
                     await pool.acquire()
@@ -349,7 +355,7 @@ async def setup(con):
             last_con = None
             cons = []
             async with self.create_pool(database='postgres',
-                                        min_size=0, max_size=1,
+                                        init_size=0, min_size=0, max_size=1,
                                         init=setup) as pool:
                 with self.assertRaises(Error):
                     await pool.acquire()
@@ -391,7 +397,7 @@ async def test_pool_auth(self):
             pool = await self.create_pool(database='postgres',
                                           user='pooluser',
                                           password='poolpassword',
-                                          min_size=5, max_size=10)
+                                          init_size=5, min_size=1, max_size=10)
 
             async def worker():
                 con = await pool.acquire()
@@ -412,7 +418,7 @@ async def worker():
     async def test_pool_handles_task_cancel_in_acquire_with_timeout(self):
         # See https://github.com/MagicStack/asyncpg/issues/547
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         async def worker():
             async with pool.acquire(timeout=100):
@@ -433,7 +439,7 @@ async def test_pool_handles_task_cancel_in_release(self):
         # Use SlowResetConnectionPool to simulate
         # the Task.cancel() and __aexit__ race.
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1,
+                                      init_size=1, min_size=1, max_size=1,
                                       connection_class=SlowResetConnection)
 
         async def worker():
@@ -454,7 +460,7 @@ async def test_pool_handles_query_cancel_in_release(self):
         # Use SlowResetConnectionPool to simulate
         # the Task.cancel() and __aexit__ race.
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1,
+                                      init_size=1, min_size=1, max_size=1,
                                       connection_class=SlowCancelConnection)
 
         async def worker():
@@ -473,7 +479,7 @@ async def worker():
 
     async def test_pool_no_acquire_deadlock(self):
         async with self.create_pool(database='postgres',
-                                    min_size=1, max_size=1,
+                                    init_size=1, min_size=1, max_size=1,
                                     max_queries=1) as pool:
 
             async def sleep_and_release():
@@ -507,7 +513,7 @@ async def test(pool):
                 cons.add(con)
 
         async with self.create_pool(
-                database='postgres', min_size=10, max_size=10,
+                database='postgres', init_size=10, min_size=1, max_size=10,
                 max_queries=1, connection_class=MyConnection,
                 statement_cache_size=3) as pool:
 
@@ -518,7 +524,9 @@ async def test(pool):
     async def test_pool_release_in_xact(self):
         """Test that Connection.reset() closes any open transaction."""
         async with self.create_pool(database='postgres',
-                                    min_size=1, max_size=1) as pool:
+                                    init_size=1,
+                                    min_size=1,
+                                    max_size=1) as pool:
             async def get_xact_id(con):
                 return await con.fetchval('select txid_current()')
 
@@ -581,7 +589,9 @@ async def test_execute_with_arg(pool):
 
         async def run(N, meth):
             async with self.create_pool(database='postgres',
-                                        min_size=5, max_size=10) as pool:
+                                        init_size=5,
+                                        min_size=0,
+                                        max_size=10) as pool:
 
                 coros = [meth(pool) for _ in range(N)]
                 res = await asyncio.gather(*coros)
@@ -608,7 +618,9 @@ async def worker(pool):
         N = 200
 
         async with self.create_pool(database='postgres',
-                                    min_size=5, max_size=10) as pool:
+                                    init_size=5,
+                                    min_size=0,
+                                    max_size=10) as pool:
 
             await pool.execute('CREATE TABLE exmany (a text, b int)')
             try:
@@ -625,7 +637,7 @@ async def worker(pool):
 
     async def test_pool_max_inactive_time_01(self):
         async with self.create_pool(
-                database='postgres', min_size=1, max_size=1,
+                database='postgres', init_size=1, min_size=0, max_size=1,
                 max_inactive_connection_lifetime=0.1) as pool:
 
             # Test that it's OK if a query takes longer time to execute
@@ -644,7 +656,7 @@ async def test_pool_max_inactive_time_01(self):
 
     async def test_pool_max_inactive_time_02(self):
         async with self.create_pool(
-                database='postgres', min_size=1, max_size=1,
+                database='postgres', init_size=1, min_size=0, max_size=1,
                 max_inactive_connection_lifetime=0.5) as pool:
 
             # Test that we have a new connection after pool not
@@ -667,7 +679,7 @@ async def test_pool_max_inactive_time_02(self):
 
     async def test_pool_max_inactive_time_03(self):
         async with self.create_pool(
-                database='postgres', min_size=1, max_size=1,
+                database='postgres', init_size=1, min_size=0, max_size=1,
                 max_inactive_connection_lifetime=1) as pool:
 
             # Test that we start counting inactive time *after*
@@ -708,7 +720,7 @@ async def worker(pool):
             N += 1
 
         async with self.create_pool(
-                database='postgres', min_size=10, max_size=30,
+                database='postgres', init_size=10, min_size=0, max_size=30,
                 max_inactive_connection_lifetime=0.1) as pool:
 
             workers = [worker(pool) for _ in range(50)]
@@ -720,7 +732,7 @@ async def test_pool_max_inactive_time_05(self):
         # Test that idle never-acquired connections abide by
         # the max inactive lifetime.
         async with self.create_pool(
-                database='postgres', min_size=2, max_size=2,
+                database='postgres', init_size=2, min_size=0, max_size=2,
                 max_inactive_connection_lifetime=0.2) as pool:
 
             self.assertIsNotNone(pool._holders[0]._con)
@@ -734,9 +746,120 @@ async def test_pool_max_inactive_time_05(self):
             # but should be closed nonetheless.
             self.assertIs(pool._holders[1]._con, None)
 
+    async def test_pool_min_size_keeps_connections_alive(self):
+        # Test that min_size prevents idle connections from being closed.
+        async with self.create_pool(
+                database='postgres', init_size=2, min_size=2, max_size=2,
+                max_inactive_connection_lifetime=0.2) as pool:
+
+            con0 = pool._holders[0]._con
+            con1 = pool._holders[1]._con
+            self.assertIsNotNone(con0)
+            self.assertIsNotNone(con1)
+
+            await asyncio.sleep(0.5)
+
+            # Connections should be kept alive because pool size == min_size.
+            self.assertIs(pool._holders[0]._con, con0)
+            self.assertIs(pool._holders[1]._con, con1)
+
+    async def test_pool_min_size_partial_keep(self):
+        # When the pool has more connections than min_size, only the excess
+        # connections should be allowed to expire; the min_size ones are kept.
+        async with self.create_pool(
+                database='postgres', init_size=3, min_size=1, max_size=3,
+                max_inactive_connection_lifetime=0.2) as pool:
+
+            # Force all 3 connections to be created by acquiring them all.
+            c1 = await pool.acquire()
+            c2 = await pool.acquire()
+            c3 = await pool.acquire()
+            await pool.release(c1)
+            await pool.release(c2)
+            await pool.release(c3)
+
+            self.assertEqual(pool.get_size(), 3)
+
+            await asyncio.sleep(0.5)
+
+            # At least min_size (1) connection must survive.
+            alive = sum(
+                1 for h in pool._holders if h._con is not None
+            )
+            self.assertGreaterEqual(alive, 1)
+
+    async def test_pool_min_size_zero_allows_full_expiry(self):
+        # When min_size=0, all idle connections are allowed to expire.
+        async with self.create_pool(
+                database='postgres', init_size=2, min_size=0, max_size=2,
+                max_inactive_connection_lifetime=0.2) as pool:
+
+            self.assertIsNotNone(pool._holders[0]._con)
+            self.assertIsNotNone(pool._holders[1]._con)
+
+            await asyncio.sleep(0.5)
+
+            self.assertIs(pool._holders[0]._con, None)
+            self.assertIs(pool._holders[1]._con, None)
+
+    async def test_pool_min_size_validation(self):
+        # init_size < min_size should raise.
+        with self.assertRaisesRegex(ValueError,
+                                    'init_size is smaller than min_size'):
+            await self.create_pool(
+                database='postgres', init_size=1, min_size=2, max_size=5)
+
+        # min_size > max_size should raise.
+        with self.assertRaisesRegex(ValueError,
+                                    'min_size is greater than max_size'):
+            await self.create_pool(
+                database='postgres', init_size=3, min_size=3, max_size=2)
+
+        # init_size > max_size should raise.
+        with self.assertRaisesRegex(ValueError,
+                                    'init_size is greater than max_size'):
+            await self.create_pool(
+                database='postgres', init_size=5, min_size=1, max_size=3)
+
+        # init_size < 0 should raise.
+        with self.assertRaisesRegex(
+                ValueError,
+                'init_size is expected to be greater or equal to zero'):
+            await self.create_pool(
+                database='postgres', init_size=-1, min_size=0, max_size=3)
+
+    async def test_pool_init_size_and_min_size_getters(self):
+        async with self.create_pool(
+                database='postgres',
+                init_size=3,
+                min_size=2,
+                max_size=5) as pool:
+            self.assertEqual(pool.get_init_size(), 3)
+            self.assertEqual(pool.get_min_size(), 2)
+            self.assertEqual(pool.get_max_size(), 5)
+            self.assertEqual(pool.get_size(), 3)
+
+    async def test_pool_min_size_reconnect_after_expiry(self):
+        # Connections kept alive by min_size should still be functional.
+        async with self.create_pool(
+                database='postgres', init_size=1, min_size=1, max_size=1,
+                max_inactive_connection_lifetime=0.2) as pool:
+
+            con_before = pool._holders[0]._con
+            self.assertIsNotNone(con_before)
+
+            await asyncio.sleep(0.5)
+
+            # Connection must still be alive due to min_size=1.
+            self.assertIs(pool._holders[0]._con, con_before)
+
+            # And it must still work.
+            result = await pool.fetchval('SELECT 42::int')
+            self.assertEqual(result, 42)
+
     async def test_pool_handles_inactive_connection_errors(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=0, max_size=1)
 
         con = await pool.acquire(timeout=POOL_NOMINAL_TIMEOUT)
 
@@ -757,10 +880,12 @@ async def test_pool_handles_inactive_connection_errors(self):
     async def test_pool_size_and_capacity(self):
         async with self.create_pool(
             database='postgres',
-            min_size=2,
+            init_size=2,
+            min_size=1,
             max_size=3,
         ) as pool:
-            self.assertEqual(pool.get_min_size(), 2)
+            self.assertEqual(pool.get_init_size(), 2)
+            self.assertEqual(pool.get_min_size(), 1)
             self.assertEqual(pool.get_max_size(), 3)
             self.assertEqual(pool.get_size(), 2)
             self.assertEqual(pool.get_idle_size(), 2)
@@ -788,7 +913,7 @@ async def test_pool_closing(self):
 
     async def test_pool_handles_transaction_exit_in_asyncgen_1(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         locals_ = {}
         exec(textwrap.dedent('''\
@@ -809,7 +934,7 @@ class MyException(Exception):
 
     async def test_pool_handles_transaction_exit_in_asyncgen_2(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         locals_ = {}
         exec(textwrap.dedent('''\
@@ -833,7 +958,7 @@ class MyException(Exception):
 
     async def test_pool_handles_asyncgen_finalization(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         locals_ = {}
         exec(textwrap.dedent('''\
@@ -854,7 +979,7 @@ class MyException(Exception):
 
     async def test_pool_close_waits_for_release(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         flag = self.loop.create_future()
         conn_released = False
@@ -877,7 +1002,7 @@ async def worker():
 
     async def test_pool_close_timeout(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         flag = self.loop.create_future()
 
@@ -896,7 +1021,7 @@ async def worker():
 
     async def test_pool_expire_connections(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         con = await pool.acquire()
         try:
@@ -909,7 +1034,7 @@ async def test_pool_expire_connections(self):
 
     async def test_pool_set_connection_args(self):
         pool = await self.create_pool(database='postgres',
-                                      min_size=1, max_size=1)
+                                      init_size=1, min_size=1, max_size=1)
 
         # Test that connection is expired on release.
         con = await pool.acquire()
@@ -951,7 +1076,12 @@ async def test_pool_set_connection_args(self):
         await pool.close()
 
     async def test_pool_init_race(self):
-        pool = self.create_pool(database='postgres', min_size=1, max_size=1)
+        pool = self.create_pool(
+            database='postgres',
+            init_size=1,
+            min_size=1,
+            max_size=1,
+        )
 
         t1 = asyncio.ensure_future(pool)
         t2 = asyncio.ensure_future(pool)
@@ -965,7 +1095,12 @@ async def test_pool_init_race(self):
         await pool.close()
 
     async def test_pool_init_and_use_race(self):
-        pool = self.create_pool(database='postgres', min_size=1, max_size=1)
+        pool = self.create_pool(
+            database='postgres',
+            init_size=1,
+            min_size=1,
+            max_size=1,
+        )
 
         pool_task = asyncio.ensure_future(pool)
         await asyncio.sleep(0)
@@ -980,7 +1115,7 @@ async def test_pool_init_and_use_race(self):
         await pool.close()
 
     async def test_pool_remote_close(self):
-        pool = await self.create_pool(min_size=1, max_size=1)
+        pool = await self.create_pool(init_size=1, min_size=1, max_size=1)
         backend_pid_fut = self.loop.create_future()
 
         async def worker():
@@ -1037,6 +1172,7 @@ async def test_full_reconnect_on_node_change_role(self):
             return
 
         pool = await self.create_pool(
+            init_size=1,
             min_size=1,
             max_size=1,
             target_session_attrs='primary'
@@ -1081,7 +1217,7 @@ async def test_standby_pool_01(self):
             with self.subTest(tasksnum=n):
                 pool = await self.create_pool(
                     database='postgres', user='postgres',
-                    min_size=5, max_size=10)
+                    init_size=5, min_size=0, max_size=10)
 
                 async def worker():
                     con = await pool.acquire()