Track re-sync servers (#170)

* Track re-sync servers

* dont change position

* Interesting

* ok pgbench tell me whats wrong

* ok

Author: levkk

integration/load_balancer/docker/primary.sh

@@ -14,4 +14,4 @@ done
 
 SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
 pushd ${SCRIPT_DIR}/../../
-RUST_LOG=debug cargo run -- --config ${SCRIPT_DIR}/pgdog.toml --users ${SCRIPT_DIR}/users.toml
+cargo run -- --config ${SCRIPT_DIR}/pgdog.toml --users ${SCRIPT_DIR}/users.toml

integration/load_balancer/docker/replica.sh

@@ -3,6 +3,9 @@
 import pytest_asyncio
 from datetime import datetime
 import json
+from time import sleep
+import random
+import asyncio
 
 @pytest_asyncio.fixture
 async def conn():
@@ -32,5 +35,48 @@ async def test_prepared_statements(conn):
             assert row[1] == "[email protected]"
             assert row[3] == json.dumps({"banned": False})
 
-    row = await conn.fetch("SELECT * FROM users WHERE id = $1", 1)
-    assert row[0][1] == "[email protected]"
+    for _ in range(3):
+        try:
+            row = await conn.fetch("SELECT * FROM users WHERE id = $1", 1)
+            assert row[0][1] == "[email protected]"
+            break
+        except:
+            # Replica lag
+            sleep(1)
+
+@pytest.mark.asyncio
+async def test_concurrent():
+    pool = await asyncpg.create_pool("postgres://postgres:[email protected]:6432/postgres")
+    tasks = []
+    for _ in range(25):
+        task = asyncio.create_task(concurrent(pool))
+        tasks.append(task)
+    for task in tasks:
+        await task
+
+async def concurrent(pool):
+    for _ in range(25):
+        async with pool.acquire() as conn:
+            i = random.randint(1, 1_000_000_000)
+            row = await conn.fetch("INSERT INTO users (id, created_at) VALUES ($1, NOW()) RETURNING id", i)
+            assert row[0][0] == i
+
+        # Read from primary
+        async with pool.acquire() as conn:
+            async with conn.transaction():
+                row = await conn.fetch("SELECT * FROM users WHERE id = $1", i)
+                assert row[0][0] == i
+
+        async with pool.acquire() as conn:
+            # Try read from replica
+            for _ in range(3):
+                try:
+                    row = await conn.fetch("SELECT * FROM users WHERE id = $1", i)
+                    assert row[0][0] == i
+                    break
+                except Exception as e:
+                    assert "list index out of range" in str(e)
+                    sleep(1)
+
+        async with pool.acquire() as conn:
+            await conn.execute("DELETE FROM users WHERE id = $1", i)

integration/load_balancer/pgdog.sh

@@ -141,7 +141,6 @@ async def test_reads_writes(engines):
 
 
 @pytest.mark.asyncio
-@pytest.mark.skip
 async def test_write_in_read(engines):
     normal = engines[0]