Репликация PostgreSQL: read-replica и маршрутизация запросов

// HikariCP + AbstractRoutingDataSource + LazyConnectionDataSourceProxy
public enum DataSourceType { MASTER, REPLICA }

@Component
public class TransactionRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return TransactionSynchronizationManager.isCurrentTransactionReadOnly()
            ? DataSourceType.REPLICA
            : DataSourceType.MASTER;
    }
}

@Configuration
public class DataSourceConfig {

    @Bean @Primary
    public DataSource routingDataSource(DataSource master, DataSource replica) {
        var routing = new TransactionRoutingDataSource();
        routing.setTargetDataSources(Map.of(
            DataSourceType.MASTER, master,
            DataSourceType.REPLICA, replica
        ));
        routing.setDefaultTargetDataSource(master);
        return new LazyConnectionDataSourceProxy(routing);
    }
}

// Использование:
@Transactional(readOnly = true)
public List<OrderView> findOrders(long customerId) {
    // уходит на реплику
}

@Transactional
public OrderId createOrder(CreateOrderCommand cmd) {
    // уходит на мастер
}

// pgxpool: два пула, выбор через контекст
type DB struct {
    Master  *pgxpool.Pool
    Replica *pgxpool.Pool
}

type ctxKey string
const readOnlyKey ctxKey = "readOnly"

func WithReadOnly(ctx context.Context) context.Context {
    return context.WithValue(ctx, readOnlyKey, true)
}

func (db *DB) Pool(ctx context.Context) *pgxpool.Pool {
    if v, ok := ctx.Value(readOnlyKey).(bool); ok && v {
        return db.Replica
    }
    return db.Master
}

// Использование:
func (r *OrderRepo) FindOrders(ctx context.Context, customerID int64) ([]Order, error) {
    rows, err := r.db.Pool(WithReadOnly(ctx)).Query(ctx,
        "SELECT id, status FROM orders WHERE customer_id = $1", customerID)
    // ...
}

func (r *OrderRepo) CreateOrder(ctx context.Context, cmd CreateOrderCmd) (int64, error) {
    var id int64
    err := r.db.Pool(ctx).QueryRow(ctx,
        "INSERT INTO orders (customer_id) VALUES ($1) RETURNING id", cmd.CustomerID,
    ).Scan(&id)
    return id, err
}

// node-postgres (pg): два Pool, выбор функцией
const pools = {
    master:  new Pool({ connectionString: process.env.DB_MASTER_URL }),
    replica: new Pool({ connectionString: process.env.DB_REPLICA_URL }),
};

function getPool(readOnly: boolean): Pool {
    return readOnly ? pools.replica : pools.master;
}

// Использование:
export async function findOrders(customerId: bigint): Promise<Order[]> {
    const { rows } = await getPool(true).query<Order>(
        'SELECT id, status FROM orders WHERE customer_id = $1',
        [customerId],
    );
    return rows;
}

export async function createOrder(cmd: CreateOrderCmd): Promise<bigint> {
    const { rows } = await getPool(false).query<{ id: bigint }>(
        'INSERT INTO orders (customer_id) VALUES ($1) RETURNING id',
        [cmd.customerId],
    );
    return rows[0].id;
}

# psycopg (v3): два пула через AsyncConnectionPool
master_pool  = AsyncConnectionPool(conninfo=MASTER_DSN, open=False)
replica_pool = AsyncConnectionPool(conninfo=REPLICA_DSN, open=False)

def get_pool(read_only: bool) -> AsyncConnectionPool:
    return replica_pool if read_only else master_pool

# Использование:
async def find_orders(customer_id: int) -> list[dict]:
    async with get_pool(read_only=True).connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT id, status FROM orders WHERE customer_id = %s",
                (customer_id,),
            )
            return await cur.fetchall()

async def create_order(customer_id: int) -> int:
    async with get_pool(read_only=False).connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "INSERT INTO orders (customer_id) VALUES (%s) RETURNING id",
                (customer_id,),
            )
            row = await cur.fetchone()
            return row[0]

@Transactional   // без readOnly=true — пойдёт на мастер
public List<Order> myOrdersFromMaster(long customerId) {
    return orderRepo.findByCustomerId(customerId);
}

// ctx без WithReadOnly — выбирается пул мастера
func (r *OrderRepo) MyOrdersFromMaster(ctx context.Context, customerID int64) ([]Order, error) {
    rows, err := r.db.Pool(ctx).Query(ctx,
        "SELECT id, status FROM orders WHERE customer_id = $1", customerID)
    // ...
}

// getPool(false) — явно мастер
export async function myOrdersFromMaster(customerId: bigint): Promise<Order[]> {
    const { rows } = await getPool(false).query<Order>(
        'SELECT id, status FROM orders WHERE customer_id = $1',
        [customerId],
    );
    return rows;
}

# read_only=False — явно мастер
async def my_orders_from_master(customer_id: int) -> list[dict]:
    async with get_pool(read_only=False).connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT id, status FROM orders WHERE customer_id = %s",
                (customer_id,),
            )
            return await cur.fetchall()

// jOOQ: INSERT ... RETURNING возвращает запись мастера
public OrderResponse createOrder(CreateOrderCommand cmd) {
    OrdersRecord saved = dsl
        .insertInto(ORDERS)
        .set(ORDERS.CUSTOMER_ID, cmd.customerId())
        .returning()
        .fetchOne();
    return OrderResponse.from(saved);
}

func (r *OrderRepo) CreateOrder(ctx context.Context, cmd CreateOrderCmd) (*Order, error) {
    var o Order
    err := r.db.Pool(ctx).QueryRow(ctx,
        `INSERT INTO orders (customer_id) VALUES ($1)
         RETURNING id, customer_id, created_at`,
        cmd.CustomerID,
    ).Scan(&o.ID, &o.CustomerID, &o.CreatedAt)
    return &o, err
}

export async function createOrder(cmd: CreateOrderCmd): Promise<Order> {
    const { rows } = await getPool(false).query<Order>(
        `INSERT INTO orders (customer_id) VALUES ($1)
         RETURNING id, customer_id, created_at`,
        [cmd.customerId],
    );
    return rows[0];
}

async def create_order(customer_id: int) -> dict:
    async with get_pool(read_only=False).connection() as conn:
        async with conn.cursor(row_factory=dict_row) as cur:
            await cur.execute(
                """INSERT INTO orders (customer_id) VALUES (%s)
                   RETURNING id, customer_id, created_at""",
                (customer_id,),
            )
            return await cur.fetchone()

String lsn = masterJdbc.queryForObject(
    "SELECT pg_current_wal_lsn()", String.class);

do {
    String replayLsn = replicaJdbc.queryForObject(
        "SELECT pg_last_wal_replay_lsn()", String.class);
    if (lsnGte(replayLsn, lsn)) break;
    Thread.sleep(50);
} while (true);

func waitForReplica(ctx context.Context, db *DB, lsn string) error {
    for {
        var replayLSN string
        err := db.Replica.QueryRow(ctx,
            "SELECT pg_last_wal_replay_lsn()").Scan(&replayLSN)
        if err != nil {
            return err
        }
        if lsnGte(replayLSN, lsn) {
            return nil
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(50 * time.Millisecond):
        }
    }
}

async function waitForReplica(lsn: string, timeoutMs = 5000): Promise<void> {
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {
        const { rows } = await getPool(true).query<{ replay: string }>(
            'SELECT pg_last_wal_replay_lsn() AS replay',
        );
        if (lsnGte(rows[0].replay, lsn)) return;
        await new Promise(r => setTimeout(r, 50));
    }
    throw new Error('replica catch-up timeout');
}

async def wait_for_replica(lsn: str, timeout: float = 5.0) -> None:
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    while loop.time() < deadline:
        async with get_pool(read_only=True).connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT pg_last_wal_replay_lsn()")
                (replay_lsn,) = await cur.fetchone()
        if lsn_gte(str(replay_lsn), lsn):
            return
        await asyncio.sleep(0.05)
    raise TimeoutError("replica catch-up timeout")

@Retryable(
    retryFor = SQLException.class,
    maxAttempts = 5,
    backoff = @Backoff(delay = 1000, multiplier = 2, random = true)
)
public OrderId createOrder(CreateOrderCommand cmd) {
    // важно: метод должен быть идемпотентным — см. пояснение ниже
}

func withRetry(ctx context.Context, maxAttempts int, fn func() error) error {
    delay := time.Second
    for attempt := range maxAttempts {
        err := fn()
        if err == nil {
            return nil
        }
        if attempt == maxAttempts-1 {
            return err
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(delay):
            delay *= 2
        }
    }
    return nil
}

async function withRetry<T>(
    fn: () => Promise<T>,
    maxAttempts = 5,
    delayMs = 1000,
): Promise<T> {
    for (let attempt = 0; attempt < maxAttempts; attempt++) {
        try {
            return await fn();
        } catch (err) {
            if (attempt === maxAttempts - 1) throw err;
            await new Promise(r => setTimeout(r, delayMs * 2 ** attempt));
        }
    }
    throw new Error('unreachable');
}

from tenacity import retry, stop_after_attempt, wait_exponential
from psycopg import OperationalError

@retry(
    retry=retry_if_exception_type(OperationalError),
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=16),
)
async def create_order(customer_id: int) -> int:
    async with get_pool(read_only=False).connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "INSERT INTO orders (customer_id) VALUES (%s) RETURNING id",
                (customer_id,),
            )
            row = await cur.fetchone()
            return row[0]