pgdog.git / summary / log / commit / refs
commit d387551d3fd46ac0acbd6f6f7a4529c8153aea7b
Author: Lev Kokotov <levkk@users.noreply.github.com>
Commit: GitHub <noreply@github.com>
Date: Wed Aug 20 21:17:11 2025 +0000
Cleanup mirror and add tests (#351)
integration/mirror/dev.sh | 5 ++
integration/mirror/ruby/.rubocop.yml | 2 +
integration/mirror/ruby/Gemfile | 7 ++
integration/mirror/ruby/Gemfile.lock | 62 ++++++++++++++++
integration/mirror/ruby/copy_spec.rb | 82 ++++++++++++++++++++++
integration/mirror/run.sh | 11 +++
.../pool/connection/mirror/buffer_with_delay.rs | 13 ++++
.../src/backend/pool/connection/mirror/handler.rs | 24 ++++++-
pgdog/src/backend/pool/connection/mirror/mod.rs | 32 +++------
.../src/backend/pool/connection/mirror/request.rs | 16 +++++
10 files changed, 231 insertions(+), 23 deletions(-)
diff --git a/integration/mirror/dev.sh b/integration/mirror/dev.sh
new file mode 100644
index 00000000..9ae67efb
--- /dev/null
+++ b/integration/mirror/dev.sh
@@ -0,0 +1,5 @@
+#!/bin/bash
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+pushd ${SCRIPT_DIR}/ruby
+bundle exec *_spec.rb
+popd
diff --git a/integration/mirror/ruby/.rubocop.yml b/integration/mirror/ruby/.rubocop.yml
new file mode 100644
index 00000000..45a9174a
--- /dev/null
+++ b/integration/mirror/ruby/.rubocop.yml
@@ -0,0 +1,2 @@
+Metrics/BlockLength:
+ Enabled: false
diff --git a/integration/mirror/ruby/Gemfile b/integration/mirror/ruby/Gemfile
new file mode 100644
index 00000000..768b5b5a
--- /dev/null
+++ b/integration/mirror/ruby/Gemfile
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+source 'https://rubygems.org'
+gem 'pg'
+gem 'rspec', '~> 3.4'
+gem 'rubocop'
+gem 'csv'
diff --git a/integration/mirror/ruby/Gemfile.lock b/integration/mirror/ruby/Gemfile.lock
new file mode 100644
index 00000000..7764a298
--- /dev/null
+++ b/integration/mirror/ruby/Gemfile.lock
@@ -0,0 +1,62 @@
+GEM
+ remote: https://rubygems.org/
+ specs:
+ ast (2.4.3)
+ csv (3.3.4)
+ diff-lcs (1.6.1)
+ json (2.12.2)
+ language_server-protocol (3.17.0.5)
+ lint_roller (1.1.0)
+ parallel (1.27.0)
+ parser (3.3.8.0)
+ ast (~> 2.4.1)
+ racc
+ pg (1.5.9)
+ prism (1.4.0)
+ racc (1.8.1)
+ rainbow (3.1.1)
+ regexp_parser (2.10.0)
+ rspec (3.13.0)
+ rspec-core (~> 3.13.0)
+ rspec-expectations (~> 3.13.0)
+ rspec-mocks (~> 3.13.0)
+ rspec-core (3.13.3)
+ rspec-support (~> 3.13.0)
+ rspec-expectations (3.13.3)
+ diff-lcs (>= 1.2.0, < 2.0)
+ rspec-support (~> 3.13.0)
+ rspec-mocks (3.13.2)
+ diff-lcs (>= 1.2.0, < 2.0)
+ rspec-support (~> 3.13.0)
+ rspec-support (3.13.2)
+ rubocop (1.76.1)
+ json (~> 2.3)
+ language_server-protocol (~> 3.17.0.2)
+ lint_roller (~> 1.1.0)
+ parallel (~> 1.10)
+ parser (>= 3.3.0.2)
+ rainbow (>= 2.2.2, < 4.0)
+ regexp_parser (>= 2.9.3, < 3.0)
+ rubocop-ast (>= 1.45.0, < 2.0)
+ ruby-progressbar (~> 1.7)
+ unicode-display_width (>= 2.4.0, < 4.0)
+ rubocop-ast (1.45.1)
+ parser (>= 3.3.7.2)
+ prism (~> 1.4)
+ ruby-progressbar (1.13.0)
+ unicode-display_width (3.1.4)
+ unicode-emoji (~> 4.0, >= 4.0.4)
+ unicode-emoji (4.0.4)
+
+PLATFORMS
+ arm64-darwin-24
+ ruby
+
+DEPENDENCIES
+ csv
+ pg
+ rspec (~> 3.4)
+ rubocop
+
+BUNDLED WITH
+ 2.6.8
diff --git a/integration/mirror/ruby/copy_spec.rb b/integration/mirror/ruby/copy_spec.rb
new file mode 100644
index 00000000..f0c2c728
--- /dev/null
+++ b/integration/mirror/ruby/copy_spec.rb
@@ -0,0 +1,82 @@
+# frozen_string_literal: true
+
+require 'rspec'
+require 'pg'
+require 'csv'
+
+describe 'mirror copy' do
+ conn = PG.connect('postgres://pgdog:pgdog@127.0.0.1:6432/pgdog')
+ mirror = PG.connect('postgres://pgdog:pgdog@127.0.0.1:6432/pgdog_mirror')
+
+ before do
+ conn.exec 'DROP TABLE IF EXISTS mirror_copy_test'
+ conn.exec 'CREATE TABLE mirror_copy_test (id BIGINT PRIMARY KEY, value VARCHAR)'
+ end
+
+ it 'can copy CSV' do
+ conn.copy_data("COPY mirror_copy_test (id, value) FROM STDIN WITH (FORMAT CSV, NULL '\\N', HEADER)") do
+ rows = [
+ %w[id value],
+ %w[1 hello@test.com],
+ [2, 'longer text in quotes'],
+ [3, nil]
+ ]
+
+ rows.each do |row|
+ conn.put_copy_data(CSV.generate_line(row, force_quotes: true, nil_value: '\\N'))
+ end
+ end
+
+ rows = conn.exec 'SELECT * FROM mirror_copy_test'
+ expect(rows.ntuples).to eq(3)
+
+ # Wait for mirror flush.
+ sleep(0.5)
+ rows = mirror.exec 'SELECT * FROM mirror_copy_test'
+ expect(rows.ntuples).to eq(3)
+ end
+
+ after do
+ conn.exec 'DROP TABLE IF EXISTS mirror_copy_test'
+ end
+end
+
+describe 'mirror crud' do
+ conn = PG.connect('postgres://pgdog:pgdog@127.0.0.1:6432/pgdog')
+ mirror = PG.connect('postgres://pgdog:pgdog@127.0.0.1:6432/pgdog_mirror')
+
+ before do
+ conn.exec 'DROP TABLE IF EXISTS mirror_crud_test'
+ conn.exec 'CREATE TABLE mirror_crud_test (id BIGINT PRIMARY KEY, value VARCHAR)'
+ conn.prepare 'insert', 'INSERT INTO mirror_crud_test VALUES ($1, $2) RETURNING *'
+ end
+
+ it 'can insert rows' do
+ results = conn.exec_prepared 'insert', [1, 'hello world']
+ expect(results.ntuples).to eq(1)
+ results = conn.exec_prepared 'insert', [2, 'apples and oranges']
+ expect(results.ntuples).to eq(1)
+
+ # Wait for mirror flush
+ sleep(0.5)
+
+ results = mirror.exec 'SELECT * FROM mirror_crud_test WHERE id = $1 AND value = $2', [1, 'hello world']
+ expect(results.ntuples).to eq(1)
+ end
+
+ it 'can update rows' do
+ conn.exec "INSERT INTO mirror_crud_test VALUES (3, 'update me')"
+ sleep(0.5)
+ result = mirror.exec 'SELECT * FROM mirror_crud_test WHERE id = 3'
+ expect(result.ntuples).to eq(1)
+ expect(result[0]['value']).to eq('update me')
+ conn.exec 'UPDATE mirror_crud_test SET value = $1 WHERE id = $2', ['updated value', 3]
+ sleep(0.5)
+ result = mirror.exec 'SELECT * FROM mirror_crud_test WHERE id = 3'
+ expect(result[0]['value']).to eq('updated value')
+ end
+
+ after do
+ conn.exec 'DROP TABLE IF EXISTS mirror_copy_test'
+ end
+end
diff --git a/integration/mirror/run.sh b/integration/mirror/run.sh
new file mode 100644
index 00000000..70970a1c
--- /dev/null
+++ b/integration/mirror/run.sh
@@ -0,0 +1,11 @@
+#!/bin/bash
+set -e
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+source ${SCRIPT_DIR}/../common.sh
+
+run_pgdog $SCRIPT_DIR
+wait_for_pgdog
+
+bash ${SCRIPT_DIR}/dev.sh
+
+stop_pgdog
diff --git a/pgdog/src/backend/pool/connection/mirror/buffer_with_delay.rs b/pgdog/src/backend/pool/connection/mirror/buffer_with_delay.rs
new file mode 100644
index 00000000..8c0541f8
--- /dev/null
+++ b/pgdog/src/backend/pool/connection/mirror/buffer_with_delay.rs
@@ -0,0 +1,13 @@
+//! Client request with duration from previous request,
+//! to simulate similar query timings on the mirror.
+
+use std::time::Duration;
+
+use crate::frontend::Buffer;
+
+/// Simulate original delay between requests.
+#[derive(Clone, Debug)]
+pub struct BufferWithDelay {
+ pub(super) delay: Duration,
+ pub(super) buffer: Buffer,
+}
diff --git a/pgdog/src/backend/pool/connection/mirror/handler.rs b/pgdog/src/backend/pool/connection/mirror/handler.rs
index fa3ab03e..339641cb 100644
--- a/pgdog/src/backend/pool/connection/mirror/handler.rs
+++ b/pgdog/src/backend/pool/connection/mirror/handler.rs
@@ -1,18 +1,35 @@
+//! Mirror client's handler.
+//!
+//! Buffers requests and simulates delay between queries.
+//!
+
use super::*;
+/// Mirror handle state.
#[derive(Debug, Clone, PartialEq, Copy)]
enum MirrorHandlerState {
+ /// Subsequent requests will be dropped until
+ /// mirror handle is flushed.
Dropping,
+ /// Requests are being buffered and will be forwarded
+ /// to the mirror when flushed.
Sending,
+ /// Mirror handle is idle.
Idle,
}
+/// Mirror handle.
#[derive(Debug)]
pub struct MirrorHandler {
+ /// Sender.
tx: Sender<MirrorRequest>,
+ /// Percentage of requests being mirrored. 0 = 0%, 1.0 = 100%.
exposure: f32,
+ /// Mirror handle state.
state: MirrorHandlerState,
+ /// Request buffer.
buffer: Vec<BufferWithDelay>,
+ /// Request timer, to simulate delays between queries.
timer: Instant,
}
@@ -22,6 +39,7 @@ impl MirrorHandler {
&self.buffer
}
+ /// Create new mirror handle with exposure.
pub fn new(tx: Sender<MirrorRequest>, exposure: f32) -> Self {
Self {
tx,
@@ -32,7 +50,10 @@ impl MirrorHandler {
}
}
- /// Maybe send request to handler.
+ /// Request the buffer to be sent to the mirror.
+ ///
+ /// Returns true if request will be sent, false otherwise.
+ ///
pub fn send(&mut self, buffer: &Buffer) -> bool {
match self.state {
MirrorHandlerState::Dropping => {
@@ -72,6 +93,7 @@ impl MirrorHandler {
}
}
+ /// Flush buffered requests to mirror.
pub fn flush(&mut self) -> bool {
if self.state == MirrorHandlerState::Dropping {
debug!("mirror transaction dropped");
diff --git a/pgdog/src/backend/pool/connection/mirror/mod.rs b/pgdog/src/backend/pool/connection/mirror/mod.rs
index 6a560498..967b28fe 100644
--- a/pgdog/src/backend/pool/connection/mirror/mod.rs
+++ b/pgdog/src/backend/pool/connection/mirror/mod.rs
@@ -1,4 +1,5 @@
-use std::ops::Deref;
+//! Client request mirroring.
+
use std::time::Duration;
use rand::{thread_rng, Rng};
@@ -20,29 +21,16 @@ use crate::frontend::Buffer;
use super::Error;
+pub mod buffer_with_delay;
pub mod handler;
-pub use handler::*;
-
-/// Simulate original delay between requests.
-#[derive(Clone, Debug)]
-pub struct BufferWithDelay {
- delay: Duration,
- buffer: Buffer,
-}
-
-#[derive(Clone, Debug)]
-pub struct MirrorRequest {
- buffer: Vec<BufferWithDelay>,
-}
-
-impl Deref for MirrorRequest {
- type Target = Vec<BufferWithDelay>;
+pub mod request;
- fn deref(&self) -> &Self::Target {
- &self.buffer
- }
-}
+pub use buffer_with_delay::*;
+pub use handler::*;
+pub use request::*;
+/// Mirror handler. One is created for each client connected
+/// to PgDog.
#[derive(Debug)]
pub struct Mirror {
/// Mirror's prepared statements. Should be similar
@@ -115,7 +103,7 @@ impl Mirror {
error!("mirror error: {}", err);
}
} else {
- debug!("mirror connection shutting down");
+ debug!("mirror client shutting down");
break;
}
}
diff --git a/pgdog/src/backend/pool/connection/mirror/request.rs b/pgdog/src/backend/pool/connection/mirror/request.rs
new file mode 100644
index 00000000..79c84441
--- /dev/null
+++ b/pgdog/src/backend/pool/connection/mirror/request.rs
@@ -0,0 +1,16 @@
+use std::ops::Deref;
+
+use super::BufferWithDelay;
+
+#[derive(Clone, Debug)]
+pub struct MirrorRequest {
+ pub(super) buffer: Vec<BufferWithDelay>,
+}
+
+impl Deref for MirrorRequest {
+ type Target = Vec<BufferWithDelay>;
+
+ fn deref(&self) -> &Self::Target {
+ &self.buffer
+ }
+}
[parent: 8c6828fc4ed8]