Markk116/livevue-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dump some extra files

Browse Source
This commit is contained in:
Markk116
2026-03-08 22:34:03 +01:00
parent ed85af4a4f
commit 2358ca29c7
4 changed files with 1461 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
maybe_sql_integration/README.md
+358
View File
@@ -0,0 +1,358 @@
# PostgreSQL Logical Replication Cache Invalidation System
A high-performance cache invalidation system that streams PostgreSQL logical replication messages, buffers changes per transaction, and emits **transaction-atomic** cache invalidation events via Tokio broadcast channels.
## Key Advantage: Transaction Atomicity
Changes are buffered until commit, so all changes in a transaction are delivered to subscribers in a single `CacheInvalidationEvent`. This guarantees:
- **No partial state**: If a transaction changes 5 rows across 3 tables, subscribers see all 5 changes together
- **Correct semantics**: Subscribers can invalidate cache atomically and notify clients in a single batch
- **Simpler logic**: No need to track partial transactions or worry about out-of-order delivery
- **Crash recovery**: Can checkpoint at commit boundary with full event
## Architecture
### Three Core Components
#### 1. **pg_wal_parser.rs** - Logical Replication Message Parser
Low-level binary protocol parser implementing the pgoutput format spec.
**Handles:**
- All PostgreSQL 14+ logical replication message types (Begin, Relation, Insert, Update, Delete, Commit, Truncate, Type, Message)
- TupleData parsing with column values (Null, UnchangedToasted, Text, Binary)
- Streaming protocol with relation OID caching
- Multi-version protocol support via graceful degradation
**Key Types:**
- `LogicalMessage` - enum of all message types with their data
- `MessageParser` - stateful parser that maintains relation metadata
- `ColumnInfo`, `RelationInfo` - schema metadata from Relation messages
#### 2. **cache_invalidation.rs** - Transaction-Aware Event Emitter
Transforms low-level replication messages into transaction-atomic high-level events.
**Handles:**
- Per-transaction event buffering (keyed by XID)
- Tracks current transaction via `Begin`/`Commit` boundary
- Accumulates Insert/Update/Delete/Truncate changes
- Emits single atomic event on Commit with all changes
- Row data extraction with column names and values
- Primary key identification (via REPLICA IDENTITY flags)
- Tokio broadcast channel emission
**Key Types:**
- `CacheInvalidationEvent` - contains XID, LSNs, timestamp, and **all changes in transaction**
- `Change` - a single table operation (Insert/Update/Delete/Truncate) with row data
- `CacheValue` - serializable column values
- `RowData` - structured row with column map and optional keys
- `TransactionBuffer` - internal state for buffering per XID
**Entry Point:**
```rust
let mut emitter = CacheInvalidationEmitter::new(tx);
emitter.process_raw(&raw_wal_bytes).await?;
```
#### 3. **example.rs** - pgwire-replication Integration
Shows how to wire everything together with the pgwire-replication crate.
**Handles:**
- Connecting to PostgreSQL logical replication
- Processing XLogData events from replication stream
- Graceful error handling and LSN checkpointing
- Example subscriber pattern for transaction-atomic cache invalidation
- Multiple concurrent subscribers
## Data Flow
```
PostgreSQL WAL (multi-statement transaction)
↓
pgwire-replication (ReplicationClient)
↓
XLogData { bytes, wal_end }
↓
MessageParser::parse_message() [binary protocol parsing]
↓
LogicalMessage enum
↓
CacheInvalidationEmitter::process_message() [buffering per XID]
↓
Begin: Create buffer for XID
Insert/Update/Delete: Append to buffer for XID
Commit: Emit CacheInvalidationEvent with all buffered changes
↓
CacheInvalidationEvent { xid, commit_lsn, changes: [Change, Change, ...] }
↓
broadcast channel → multiple subscribers receive transaction atomically
```
## Example Transaction Flow
PostgreSQL executes:
```sql
BEGIN;
INSERT INTO users (id, name) VALUES (1, 'alice');
INSERT INTO posts (id, user_id, title) VALUES (100, 1, 'Hello');
UPDATE posts SET views = 1 WHERE id = 100;
COMMIT;
```
Replication protocol sends:
```
Begin XID=5000 LSN=0/123456
Relation OID=16384 name="users"
Relation OID=16385 name="posts"
Insert rel=16384 [1, 'alice']
Insert rel=16385 [100, 1, 'Hello']
Update rel=16385 key=[100] new=[100, 1, 'Hello', 1]
Commit XID=5000 LSN=0/789ABC
```
Emitter produces (single event):
```rust
CacheInvalidationEvent {
xid: 5000,
commit_lsn: 0x789ABC,
end_lsn: 0x789ABC,
timestamp: 1234567890,
changes: [
Change {
table: "users",
schema: "public",
operation: Insert,
row_data: Some({id: 1, name: 'alice'})
},
Change {
table: "posts",
schema: "public",
operation: Insert,
row_data: Some({id: 100, user_id: 1, title: 'Hello'})
},
Change {
table: "posts",
schema: "public",
operation: Update,
row_data: Some({
columns: {id: 100, user_id: 1, title: 'Hello', views: 1},
key_values: Some({id: 100})
})
}
]
}
```
Subscribers receive this single event and can:
- Invalidate cache for users:1, posts:100 atomically
- Notify WebSocket clients in one batch
- Checkpoint the LSN
- Update metrics
## Protocol Details
### Message Format (from PostgreSQL docs)
Each logical replication message starts with a type byte:
| Type | Meaning | When |
|------|---------|------|
| B | Begin | Transaction start (1 per txn) |
| R | Relation | Schema definition (1+ per txn) |
| I | Insert | New row written to table |
| U | Update | Row modified (includes old/new) |
| D | Delete | Row removed (includes key/old) |
| C | Commit | Transaction committed (1 per txn) |
| T | Truncate | Table truncated |
| Y | Type | Custom data type definition |
| M | Message | Custom pg_logical_emit_message() |
### Transaction Boundaries
- **Begin** message: `XID` field identifies the transaction
- **Changes** (Insert/Update/Delete/Truncate): Include same `XID`
- **Commit** message: Completes the transaction (no XID in Commit itself)
The emitter tracks `current_xid` to match Commit to its Begin, ensuring all changes are buffered and emitted together.
## Setup & Usage
### PostgreSQL Setup (one-time)
```sql
-- Enable logical replication
ALTER SYSTEM SET wal_level = logical;
SELECT pg_reload_conf();
-- Create replication slot
SELECT * FROM pg_create_logical_replication_slot('my_slot', 'pgoutput');
-- Create publication
CREATE PUBLICATION my_pub FOR TABLE users, posts, comments;
```
### Rust Code Integration
```rust
use tokio::sync::broadcast;
use cache_invalidation::CacheInvalidationEmitter;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let (tx, mut rx) = broadcast::channel(10000);
// Spawn WAL listener (uses pgwire-replication)
tokio::spawn(async move {
let mut client = pgwire_replication::ReplicationClient::connect(config).await?;
let mut emitter = CacheInvalidationEmitter::new(tx);
while let Some(event) = client.recv().await? {
match event {
pgwire_replication::ReplicationEvent::XLogData { data, wal_end, .. } => {
// Process all messages in data; buffered per transaction
emitter.process_raw(&data).await?;
client.update_applied_lsn(wal_end);
}
_ => {}
}
}
Ok::<(), Box<dyn std::error::Error>>(())
});
// Subscribe to cache invalidation events
loop {
match rx.recv().await {
Ok(event) => {
println!("Transaction XID={} committed with {} changes",
event.xid, event.changes.len());
// Invalidate cache for all changes atomically
for change in &event.changes {
match &change.operation {
Operation::Insert => {
// cache.delete_by_table(change.table)
}
Operation::Update => {
if let Some(row) = &change.row_data {
if let Some(keys) = &row.key_values {
// cache.delete(change.table, keys)
}
}
}
Operation::Delete => {
if let Some(row) = &change.row_data {
if let Some(keys) = &row.key_values {
// cache.delete(change.table, keys)
}
}
}
Operation::Truncate => {
// cache.clear_table(change.table)
}
}
}
// Notify WebSocket clients once, with all changes
// websocket_broadcast(event)
}
Err(broadcast::error::RecvError::Lagged(_)) => {
eprintln!("Subscriber lagged, reloading cache from DB");
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
Ok(())
}
```
## Performance Characteristics
- **Zero-copy parsing**: Uses `bytes::BytesMut` with in-place operations
- **Async-first**: Compatible with Tokio task scheduling
- **Streaming**: Processes messages as they arrive, no buffering of raw data
- **Transaction buffering**: Per-XID HashMap (typically 1-10 concurrent txns)
- **Memory**: Relation metadata cached by OID + per-txn change buffer
- **LSN tracking**: Explicit control for checkpointing and crash recovery
## Typical Latencies
- PostgreSQL writes change to WAL: < 1ms
- WAL transmitted to client: network latency (localhost ~0.1ms)
- Message parsing: ~10-100µs per message
- Buffering: ~1µs per change
- Commit event emission: ~10-100µs
- **End-to-end (localhost)**: ~1-5ms from commit to subscriber notification
## Transaction Buffering Internals
```rust
// On Begin: Create buffer
txn_buffers.insert(xid, TransactionBuffer { xid, changes: Vec::new() });
current_xid = Some(xid);
// On Insert/Update/Delete/Truncate: Append to buffer
if let Some(buf) = txn_buffers.get_mut(&xid) {
buf.changes.push(change);
}
// On Commit: Emit and clean up
if let Some(xid) = current_xid.take() {
emit_transaction(xid, commit_lsn, end_lsn, timestamp)?;
txn_buffers.remove(&xid);
}
```
## Testing
Each module has tests. Run with:
```bash
cargo test
```
## Caveats & Notes
### What This Doesn't Do
- Custom output plugins (uses standard pgoutput only)
- Two-phase commits (v3+) would need extended buffering
- Concurrent transaction correlation (assumes sequential Begins/Commits)
### REPLICA IDENTITY Configuration
For UPDATE/DELETE to include key information:
```sql
ALTER TABLE users REPLICA IDENTITY USING INDEX users_pkey;
-- or for full old row:
ALTER TABLE users REPLICA IDENTITY FULL;
```
Default is PRIMARY KEY, which is usually what you want.
### Slot Management
Replication slots persist until explicitly dropped. If the stream falls behind, the WAL won't be deleted and disk will fill up:
```sql
-- Monitor slot lag
SELECT slot_name, active, restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots;
-- Drop a slot when done
SELECT pg_drop_replication_slot('my_slot');
```
## Future Enhancements
1. **Metrics**: Instrument parsing and emit latency histograms
2. **Backpressure**: Honor broadcast channel capacity
3. **Checkpointing**: Persist LSN to disk/Redis for recovery
4. **Filtering**: Table/operation-level subscription filtering
5. **Parallel streams**: Support multiple concurrent WAL consumers
## References
- [PostgreSQL Logical Replication Protocol](https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html)
- [pgoutput Output Plugin](https://www.postgresql.org/docs/current/sql-createpublication.html)
- [pgwire-replication Crate](https://crates.io/crates/pgwire-replication)
- [Tokio Broadcast Channel](https://tokio.rs/tokio/tutorial/select#broadcast)
maybe_sql_integration/cache_invalidation.rs
+366
View File
@@ -0,0 +1,366 @@
use crate::pg_wal_parser::{ColumnValue, LogicalMessage, MessageParser, RelationInfo, TupleData};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use tokio::sync::broadcast;
/// High-level cache invalidation event emitted at transaction commit
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheInvalidationEvent {
/// Transaction ID from PostgreSQL
pub xid: u32,
/// Commit LSN/watermark for checkpoint
pub commit_lsn: u64,
/// End LSN of the transaction
pub end_lsn: u64,
/// Timestamp of commit (microseconds since PG epoch)
pub timestamp: i64,
/// All operations in this transaction, emitted atomically
pub changes: Vec<Change>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Change {
/// Table name that changed
pub table: String,
/// Schema/namespace
pub schema: String,
/// Operation type
pub operation: Operation,
/// Row data if available
pub row_data: Option<RowData>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum Operation {
Insert,
Update,
Delete,
Truncate,
}
/// Simplified row representation for broadcasting
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RowData {
/// Mapping of column name -> value
pub columns: HashMap<String, CacheValue>,
/// Primary key values if available
pub key_values: Option<HashMap<String, CacheValue>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CacheValue {
Null,
String(String),
Binary(Vec<u8>),
UnchangedToasted,
}
/// Transaction buffer for a single transaction
struct TransactionBuffer {
xid: u32,
changes: Vec<Change>,
}
pub struct CacheInvalidationEmitter {
parser: MessageParser,
tx: broadcast::Sender<CacheInvalidationEvent>,
/// Buffers for active transactions, keyed by XID
txn_buffers: HashMap<u32, TransactionBuffer>,
/// Track the current transaction being processed
current_xid: Option<u32>,
}
impl CacheInvalidationEmitter {
pub fn new(tx: broadcast::Sender<CacheInvalidationEvent>) -> Self {
Self {
parser: MessageParser::new(),
tx,
txn_buffers: HashMap::new(),
current_xid: None,
}
}
/// Process a single logical replication message
/// Events are buffered until Commit, then emitted atomically
pub async fn process_message(&mut self, msg: LogicalMessage) -> Result<(), String> {
match msg {
LogicalMessage::Begin { xid, .. } => {
// Start buffering for this transaction
self.current_xid = Some(xid);
self.txn_buffers.insert(xid, TransactionBuffer {
xid,
changes: Vec::new(),
});
Ok(())
}
LogicalMessage::Relation(_) => {
// Relation updates internal state via parser
Ok(())
}
LogicalMessage::Insert {
xid,
rel_oid,
new_tuple,
} => {
let change = self.build_insert_change(rel_oid, new_tuple)?;
self.buffer_change(xid, change);
Ok(())
}
LogicalMessage::Update {
xid,
rel_oid,
key_tuple,
old_tuple: _,
new_tuple,
} => {
let change = self.build_update_change(rel_oid, new_tuple, key_tuple)?;
self.buffer_change(xid, change);
Ok(())
}
LogicalMessage::Delete {
xid,
rel_oid,
key_tuple,
old_tuple,
} => {
let change = self.build_delete_change(rel_oid, key_tuple.or(old_tuple))?;
self.buffer_change(xid, change);
Ok(())
}
LogicalMessage::Commit {
commit_lsn,
end_lsn,
timestamp,
..
} => {
// Emit the buffered transaction atomically
if let Some(xid) = self.current_xid.take() {
self.emit_transaction(xid, commit_lsn, end_lsn, timestamp)?;
}
Ok(())
}
LogicalMessage::Truncate {
xid,
rel_oids,
..
} => {
for oid in rel_oids {
let change = self.build_truncate_change(oid)?;
self.buffer_change(xid, change);
}
Ok(())
}
LogicalMessage::Type { .. } => {
// Type information, no action needed
Ok(())
}
LogicalMessage::Message { .. } => {
// Custom messages, can be ignored for cache invalidation
Ok(())
}
}
}
fn buffer_change(&mut self, xid: u32, change: Change) {
if let Some(buf) = self.txn_buffers.get_mut(&xid) {
buf.changes.push(change);
}
}
fn build_insert_change(&self, rel_oid: u32, new_tuple: TupleData) -> Result<Change, String> {
let relation = self
.parser
.get_relation(rel_oid)
.ok_or_else(|| format!("Relation OID {} not found", rel_oid))?;
let row_data = self.build_row_data(relation, &new_tuple)?;
Ok(Change {
table: relation.name.clone(),
schema: relation.namespace.clone(),
operation: Operation::Insert,
row_data: Some(row_data),
})
}
fn build_update_change(
&self,
rel_oid: u32,
new_tuple: TupleData,
key_tuple: Option<TupleData>,
) -> Result<Change, String> {
let relation = self
.parser
.get_relation(rel_oid)
.ok_or_else(|| format!("Relation OID {} not found", rel_oid))?;
let mut row_data = self.build_row_data(relation, &new_tuple)?;
// If we have a key tuple, extract key values
if let Some(key) = key_tuple {
let key_values = self.extract_key_values(relation, &key)?;
row_data.key_values = Some(key_values);
}
Ok(Change {
table: relation.name.clone(),
schema: relation.namespace.clone(),
operation: Operation::Update,
row_data: Some(row_data),
})
}
fn build_delete_change(
&self,
rel_oid: u32,
key_or_old: Option<TupleData>,
) -> Result<Change, String> {
let relation = self
.parser
.get_relation(rel_oid)
.ok_or_else(|| format!("Relation OID {} not found", rel_oid))?;
let mut row_data = None;
if let Some(tuple) = key_or_old {
let mut data = self.build_row_data(relation, &tuple)?;
let key_values = self.extract_key_values(relation, &tuple)?;
data.key_values = Some(key_values);
row_data = Some(data);
}
Ok(Change {
table: relation.name.clone(),
schema: relation.namespace.clone(),
operation: Operation::Delete,
row_data,
})
}
fn build_truncate_change(&self, rel_oid: u32) -> Result<Change, String> {
let relation = self
.parser
.get_relation(rel_oid)
.ok_or_else(|| format!("Relation OID {} not found", rel_oid))?;
Ok(Change {
table: relation.name.clone(),
schema: relation.namespace.clone(),
operation: Operation::Truncate,
row_data: None,
})
}
fn build_row_data(&self, relation: &RelationInfo, tuple: &TupleData) -> Result<RowData, String> {
let mut columns = HashMap::new();
for (i, value) in tuple.values.iter().enumerate() {
if i >= relation.columns.len() {
return Err(format!(
"Tuple has more values than relation has columns: {} vs {}",
tuple.values.len(),
relation.columns.len()
));
}
let col = &relation.columns[i];
let cache_value = match value {
ColumnValue::Null => CacheValue::Null,
ColumnValue::UnchangedToasted => CacheValue::UnchangedToasted,
ColumnValue::Text(s) => CacheValue::String(s.clone()),
ColumnValue::Binary(b) => CacheValue::Binary(b.clone()),
};
columns.insert(col.name.clone(), cache_value);
}
Ok(RowData {
columns,
key_values: None,
})
}
fn extract_key_values(
&self,
relation: &RelationInfo,
key_tuple: &TupleData,
) -> Result<HashMap<String, CacheValue>, String> {
let mut key_values = HashMap::new();
for (i, value) in key_tuple.values.iter().enumerate() {
if i >= relation.columns.len() {
break; // Key might have fewer columns
}
let col = &relation.columns[i];
if col.flags & 1 != 0 {
// This column is part of the key
let cache_value = match value {
ColumnValue::Null => CacheValue::Null,
ColumnValue::UnchangedToasted => CacheValue::UnchangedToasted,
ColumnValue::Text(s) => CacheValue::String(s.clone()),
ColumnValue::Binary(b) => CacheValue::Binary(b.clone()),
};
key_values.insert(col.name.clone(), cache_value);
}
}
Ok(key_values)
}
/// Emit a buffered transaction atomically
fn emit_transaction(&mut self, xid: u32, commit_lsn: u64, end_lsn: u64, timestamp: i64) -> Result<(), String> {
if let Some(buf) = self.txn_buffers.remove(&xid) {
let event = CacheInvalidationEvent {
xid,
commit_lsn,
end_lsn,
timestamp,
changes: buf.changes,
};
self.tx.send(event).ok(); // ignore if no receivers
}
Ok(())
}
/// Parse raw replication data bytes and process messages
/// This is the main entry point from pgwire-replication
pub async fn process_raw(&mut self, data: &[u8]) -> Result<(), String> {
let mut buf = bytes::BytesMut::from(&data[..]);
while !buf.is_empty() {
if let Some(msg) = self.parser.parse_message(&mut buf)? {
self.process_message(msg).await?;
}
}
Ok(())
}
/// Get statistics about buffered transactions
pub fn buffered_transaction_count(&self) -> usize {
self.txn_buffers.len()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_emitter_creation() {
let (tx, _rx) = broadcast::channel(100);
let _emitter = CacheInvalidationEmitter::new(tx);
}
#[test]
fn test_transaction_buffering() {
let (tx, _rx) = broadcast::channel(100);
let emitter = CacheInvalidationEmitter::new(tx);
// Changes should be buffered per transaction until commit
assert_eq!(emitter.buffered_transaction_count(), 0);
}
}
maybe_sql_integration/example.rs
+259
View File
@@ -0,0 +1,259 @@
/// Example integration of pgwire-replication with cache invalidation
///
/// This shows how to:
/// 1. Connect to PostgreSQL logical replication stream
/// 2. Parse logical messages
/// 3. Buffer changes per transaction
/// 4. Emit all changes atomically at commit via Tokio broadcast
/// 5. Subscribe to events for client notification
use tokio::sync::broadcast;
use std::time::Duration;
// These would be your actual modules
mod pg_wal_parser;
mod cache_invalidation;
use cache_invalidation::{CacheInvalidationEvent, CacheInvalidationEmitter};
/// Example configuration
pub struct ReplicationConfig {
pub pg_host: String,
pub pg_port: u16,
pub pg_user: String,
pub pg_password: String,
pub pg_database: String,
pub replication_slot: String,
pub publication: String,
pub start_lsn: String, // e.g. "0/0" or "0/16B6C50"
}
/// Main replication stream handler
pub struct WalStreamListener {
config: ReplicationConfig,
tx: broadcast::Sender<CacheInvalidationEvent>,
}
impl WalStreamListener {
pub fn new(
config: ReplicationConfig,
tx: broadcast::Sender<CacheInvalidationEvent>,
) -> Self {
Self { config, tx }
}
/// Start the replication stream listener
/// In production, this would use pgwire-replication::ReplicationClient
pub async fn start(self) -> Result<(), Box<dyn std::error::Error>> {
// This is a sketch - actual implementation would use pgwire-replication
// 1. Create replication client
// let mut client = pgwire_replication::ReplicationClient::connect(
// pgwire_replication::ReplicationConfig {
// host: self.config.pg_host.into(),
// port: self.config.pg_port,
// user: self.config.pg_user.into(),
// password: self.config.pg_password.into(),
// database: self.config.pg_database.into(),
// slot: self.config.replication_slot.into(),
// publication: self.config.publication.into(),
// start_lsn: pgwire_replication::Lsn::parse(&self.config.start_lsn)?,
// status_interval: Duration::from_secs(10),
// idle_wakeup_interval: Duration::from_secs(10),
// ..Default::default()
// },
// ).await?;
// 2. Create emitter with transaction buffering
let mut emitter = CacheInvalidationEmitter::new(self.tx);
// 3. Process stream
// loop {
// match client.recv().await? {
// Some(pgwire_replication::ReplicationEvent::XLogData { data, wal_end, .. }) => {
// // Parse and buffer changes per transaction
// emitter.process_raw(&data).await?;
//
// // Update LSN checkpoint after processing
// client.update_applied_lsn(wal_end);
// }
// Some(pgwire_replication::ReplicationEvent::KeepAlive { .. }) => {
// // Server keepalive, continue
// }
// Some(pgwire_replication::ReplicationEvent::StoppedAt { .. }) => {
// // Stream ended gracefully
// break;
// }
// Some(_) => {}
// None => break,
// }
// }
Ok(())
}
}
/// Example client that subscribes to cache invalidation events
/// All changes in a transaction are emitted atomically on commit
pub async fn example_subscriber(
mut rx: broadcast::Receiver<CacheInvalidationEvent>,
) -> Result<(), Box<dyn std::error::Error>> {
loop {
match rx.recv().await {
Ok(event) => {
println!(
"Transaction committed: XID={} LSN={:016X}-{:016X} timestamp={} changes={}",
event.xid, event.commit_lsn, event.end_lsn, event.timestamp, event.changes.len()
);
// All changes in this transaction are now available atomically
for change in &event.changes {
match &change.operation {
cache_invalidation::Operation::Insert => {
println!(" INSERT {}.{}", change.schema, change.table);
}
cache_invalidation::Operation::Update => {
println!(" UPDATE {}.{}", change.schema, change.table);
}
cache_invalidation::Operation::Delete => {
println!(" DELETE {}.{}", change.schema, change.table);
}
cache_invalidation::Operation::Truncate => {
println!(" TRUNCATE {}.{}", change.schema, change.table);
}
}
if let Some(row) = &change.row_data {
if let Some(keys) = &row.key_values {
print!(" Keys: ");
for (k, v) in keys {
print!("{}={:?} ", k, v);
}
println!();
}
}
}
// In a real system, you would:
// 1. Invalidate all cache entries for this transaction atomically
// 2. Notify WebSocket clients (single transaction message)
// 3. Store event for audit/replication trail
// 4. Update Prometheus metrics
// 5. Persist LSN for crash recovery
}
Err(broadcast::error::RecvError::Lagged(_)) => {
println!("Warning: Cache invalidation queue lagged, reloading cache from DB");
}
Err(broadcast::error::RecvError::Closed) => {
println!("Cache invalidation stream closed");
break;
}
}
}
Ok(())
}
/// Example showing transaction atomicity
/// Demonstrates that a multi-table transaction is delivered as one event
async fn example_transaction_atomicity() {
println!("Example: Multi-table transaction");
println!();
println!("BEGIN");
println!(" INSERT INTO users (id, name) VALUES (1, 'alice')");
println!(" INSERT INTO posts (id, user_id, title) VALUES (100, 1, 'Hello')");
println!(" INSERT INTO comments (id, post_id, text) VALUES (1000, 100, 'Nice!')");
println!("COMMIT LSN=0/1234567 XID=5000");
println!();
println!("Result: Single CacheInvalidationEvent with 3 changes");
println!(" - Change[0]: users.INSERT");
println!(" - Change[1]: posts.INSERT");
println!(" - Change[2]: comments.INSERT");
println!();
println!("Subscribers receive all 3 changes atomically,");
println!("so they can update cache without partial state.");
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// In a real application, you might:
// 1. Create broadcast channel for cache events (large capacity for bursty writes)
let (tx, rx) = broadcast::channel::<CacheInvalidationEvent>(10000);
// 2. Spawn WAL stream listener
let config = ReplicationConfig {
pg_host: "127.0.0.1".to_string(),
pg_port: 5432,
pg_user: "postgres".to_string(),
pg_password: "postgres".to_string(),
pg_database: "mydb".to_string(),
replication_slot: "my_slot".to_string(),
publication: "my_publication".to_string(),
start_lsn: "0/0".to_string(),
};
let listener = WalStreamListener::new(config, tx);
tokio::spawn(async move {
if let Err(e) = listener.start().await {
eprintln!("Replication stream error: {}", e);
}
});
// 3. Spawn multiple subscribers (e.g., for different cache layers)
let rx_cache = tx.subscribe();
tokio::spawn(async move {
if let Err(e) = example_subscriber(rx_cache).await {
eprintln!("Cache subscriber error: {}", e);
}
});
let rx_analytics = tx.subscribe();
tokio::spawn(async move {
// Another subscriber could log for analytics
let mut rx = rx_analytics;
loop {
if let Ok(event) = rx.recv().await {
// Track metrics, e.g., changes per table per second
eprintln!("Metric: {} changes in XID {}", event.changes.len(), event.xid);
} else {
break;
}
}
});
// 4. Your application continues in the background
// The cache invalidation events flow continuously
// Keep main alive
example_transaction_atomicity().await;
tokio::signal::ctrl_c().await?;
println!("Shutting down");
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_broadcast_channel() {
let (tx, mut rx) = broadcast::channel::<CacheInvalidationEvent>(10);
let event = CacheInvalidationEvent {
xid: 1000,
commit_lsn: 0x123456,
end_lsn: 0x654321,
timestamp: 1_000_000,
changes: vec![],
};
tx.send(event.clone()).unwrap();
let received = rx.recv().await.unwrap();
assert_eq!(received.xid, 1000);
assert_eq!(received.changes.len(), 0);
}
}
maybe_sql_integration/pg_wal_parser.rs
+478
View File
@@ -0,0 +1,478 @@
use bytes::{Buf, BytesMut};
use std::collections::HashMap;
/// PostgreSQL logical replication message parser
/// Handles pgoutput protocol messages from replication streams
#[derive(Debug, Clone)]
pub struct ColumnInfo {
pub flags: u8, // 1 if part of key, 0 otherwise
pub name: String,
pub type_oid: u32,
pub type_modifier: i32,
}
#[derive(Debug, Clone)]
pub struct RelationInfo {
pub oid: u32,
pub namespace: String,
pub name: String,
pub replica_identity: u8,
pub columns: Vec<ColumnInfo>,
}
#[derive(Debug, Clone)]
pub enum ColumnValue {
Null,
UnchangedToasted,
Text(String),
Binary(Vec<u8>),
}
#[derive(Debug)]
pub struct TupleData {
pub values: Vec<ColumnValue>,
}
#[derive(Debug)]
pub enum LogicalMessage {
Begin {
final_lsn: u64,
timestamp: i64,
xid: u32,
},
Relation(RelationInfo),
Insert {
xid: u32,
rel_oid: u32,
new_tuple: TupleData,
},
Update {
xid: u32,
rel_oid: u32,
key_tuple: Option<TupleData>,
old_tuple: Option<TupleData>,
new_tuple: TupleData,
},
Delete {
xid: u32,
rel_oid: u32,
key_tuple: Option<TupleData>,
old_tuple: Option<TupleData>,
},
Commit {
flags: u8,
commit_lsn: u64,
end_lsn: u64,
timestamp: i64,
},
Type {
xid: u32,
type_oid: u32,
namespace: String,
type_name: String,
},
Truncate {
xid: u32,
num_relations: u32,
options: u8,
rel_oids: Vec<u32>,
},
Message {
xid: u32,
flags: u8,
lsn: u64,
prefix: String,
content: Vec<u8>,
},
}
pub struct MessageParser {
/// Map of relation OID -> RelationInfo for decoding tuples
relations: HashMap<u32, RelationInfo>,
/// Map of type OID -> type name for potential future use
types: HashMap<u32, (String, String)>,
}
impl MessageParser {
pub fn new() -> Self {
Self {
relations: HashMap::new(),
types: HashMap::new(),
}
}
/// Parse a single logical replication message from bytes
/// Assumes input starts with the message type byte
pub fn parse_message(&mut self, input: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
if input.is_empty() {
return Ok(None);
}
let msg_type = input[0] as char;
match msg_type {
'B' => self.parse_begin(input),
'I' => self.parse_insert(input),
'U' => self.parse_update(input),
'D' => self.parse_delete(input),
'C' => self.parse_commit(input),
'R' => {
self.parse_relation(input)?;
Ok(None) // Relation messages are stored internally
}
'Y' => self.parse_type(input),
'T' => self.parse_truncate(input),
'M' => self.parse_message_msg(input),
't' => Err("Tuple data outside of Insert/Update/Delete (unexpected)".to_string()),
_ => Err(format!("Unknown message type: {} (byte: {})", msg_type, input[0])),
}
}
fn read_string(buf: &mut BytesMut) -> Result<String, String> {
// PostgreSQL strings are null-terminated
let mut vec = Vec::new();
loop {
if buf.is_empty() {
return Err("EOF while reading string".to_string());
}
let byte = buf.get_u8();
if byte == 0 {
break;
}
vec.push(byte);
}
String::from_utf8(vec).map_err(|e| format!("Invalid UTF-8 in string: {}", e))
}
fn read_i64(buf: &mut BytesMut) -> Result<i64, String> {
if buf.len() < 8 {
return Err("Not enough bytes for i64".to_string());
}
Ok(buf.get_i64())
}
fn read_u64(buf: &mut BytesMut) -> Result<u64, String> {
if buf.len() < 8 {
return Err("Not enough bytes for u64".to_string());
}
Ok(buf.get_u64())
}
fn read_i32(buf: &mut BytesMut) -> Result<i32, String> {
if buf.len() < 4 {
return Err("Not enough bytes for i32".to_string());
}
Ok(buf.get_i32())
}
fn read_u32(buf: &mut BytesMut) -> Result<u32, String> {
if buf.len() < 4 {
return Err("Not enough bytes for u32".to_string());
}
Ok(buf.get_u32())
}
fn read_i16(buf: &mut BytesMut) -> Result<i16, String> {
if buf.len() < 2 {
return Err("Not enough bytes for i16".to_string());
}
Ok(buf.get_i16())
}
fn read_u8(buf: &mut BytesMut) -> Result<u8, String> {
if buf.is_empty() {
return Err("Not enough bytes for u8".to_string());
}
Ok(buf.get_u8())
}
fn parse_begin(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'B'
let final_lsn = Self::read_u64(buf)?;
let timestamp = Self::read_i64(buf)?;
let xid = Self::read_u32(buf)?;
Ok(Some(LogicalMessage::Begin {
final_lsn,
timestamp,
xid,
}))
}
fn parse_commit(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'C'
let flags = Self::read_u8(buf)?;
let commit_lsn = Self::read_u64(buf)?;
let end_lsn = Self::read_u64(buf)?;
let timestamp = Self::read_i64(buf)?;
Ok(Some(LogicalMessage::Commit {
flags,
commit_lsn,
end_lsn,
timestamp,
}))
}
fn parse_relation(&mut self, buf: &mut BytesMut) -> Result<(), String> {
buf.get_u8(); // 'R'
// xid only present for streamed transactions (version 2+), skip for now
let rel_oid = Self::read_u32(buf)?;
let namespace = Self::read_string(buf)?;
let name = Self::read_string(buf)?;
let replica_identity = Self::read_u8(buf)?;
let num_columns = Self::read_i16(buf)? as usize;
let mut columns = Vec::with_capacity(num_columns);
for _ in 0..num_columns {
let flags = Self::read_u8(buf)?;
let col_name = Self::read_string(buf)?;
let type_oid = Self::read_u32(buf)?;
let type_modifier = Self::read_i32(buf)?;
columns.push(ColumnInfo {
flags,
name: col_name,
type_oid,
type_modifier,
});
}
let relation = RelationInfo {
oid: rel_oid,
namespace,
name,
replica_identity,
columns,
};
self.relations.insert(rel_oid, relation);
Ok(())
}
fn parse_insert(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'I'
let xid = Self::read_u32(buf)?;
let rel_oid = Self::read_u32(buf)?;
// Next byte must be 'N' for new tuple
let marker = Self::read_u8(buf)?;
if marker != b'N' {
return Err(format!("Expected 'N' in INSERT, got: {}", marker as char));
}
let new_tuple = self.parse_tuple_data(buf, rel_oid)?;
Ok(Some(LogicalMessage::Insert {
xid,
rel_oid,
new_tuple,
}))
}
fn parse_update(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'U'
let xid = Self::read_u32(buf)?;
let rel_oid = Self::read_u32(buf)?;
let mut key_tuple = None;
let mut old_tuple = None;
// Optional K or O markers
if !buf.is_empty() {
match buf[0] as char {
'K' => {
buf.get_u8(); // consume 'K'
key_tuple = Some(self.parse_tuple_data(buf, rel_oid)?);
}
'O' => {
buf.get_u8(); // consume 'O'
old_tuple = Some(self.parse_tuple_data(buf, rel_oid)?);
}
_ => {} // neither K nor O, proceed to N
}
}
let marker = Self::read_u8(buf)?;
if marker != b'N' {
return Err(format!("Expected 'N' in UPDATE, got: {}", marker as char));
}
let new_tuple = self.parse_tuple_data(buf, rel_oid)?;
Ok(Some(LogicalMessage::Update {
xid,
rel_oid,
key_tuple,
old_tuple,
new_tuple,
}))
}
fn parse_delete(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'D'
let xid = Self::read_u32(buf)?;
let rel_oid = Self::read_u32(buf)?;
let mut key_tuple = None;
let mut old_tuple = None;
// Either K or O (but not both)
if !buf.is_empty() {
match buf[0] as char {
'K' => {
buf.get_u8(); // consume 'K'
key_tuple = Some(self.parse_tuple_data(buf, rel_oid)?);
}
'O' => {
buf.get_u8(); // consume 'O'
old_tuple = Some(self.parse_tuple_data(buf, rel_oid)?);
}
_ => return Err(format!("Expected K or O in DELETE, got: {}", buf[0] as char)),
}
}
Ok(Some(LogicalMessage::Delete {
xid,
rel_oid,
key_tuple,
old_tuple,
}))
}
fn parse_type(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'Y'
let type_oid = Self::read_u32(buf)?;
let namespace = Self::read_string(buf)?;
let type_name = Self::read_string(buf)?;
let xid = 0; // Type doesn't have xid in base protocol
self.types
.insert(type_oid, (namespace.clone(), type_name.clone()));
Ok(Some(LogicalMessage::Type {
xid,
type_oid,
namespace,
type_name,
}))
}
fn parse_truncate(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'T'
let xid = Self::read_u32(buf)?;
let num_relations = Self::read_u32(buf)?;
let options = Self::read_u8(buf)?;
let mut rel_oids = Vec::with_capacity(num_relations as usize);
for _ in 0..num_relations {
rel_oids.push(Self::read_u32(buf)?);
}
Ok(Some(LogicalMessage::Truncate {
xid,
num_relations,
options,
rel_oids,
}))
}
fn parse_message_msg(&mut self, buf: &mut BytesMut) -> Result<Option<LogicalMessage>, String> {
buf.get_u8(); // 'M'
let xid = Self::read_u32(buf)?;
let flags = Self::read_u8(buf)?;
let lsn = Self::read_u64(buf)?;
let prefix = Self::read_string(buf)?;
let content_len = Self::read_i32(buf)? as usize;
if buf.len() < content_len {
return Err(format!(
"Not enough bytes for message content: {} < {}",
buf.len(),
content_len
));
}
let content = buf.split_to(content_len).to_vec();
Ok(Some(LogicalMessage::Message {
xid,
flags,
lsn,
prefix,
content,
}))
}
fn parse_tuple_data(&self, buf: &mut BytesMut, rel_oid: u32) -> Result<TupleData, String> {
let num_columns = Self::read_i16(buf)? as usize;
let mut values = Vec::with_capacity(num_columns);
let relation = self
.relations
.get(&rel_oid)
.ok_or_else(|| format!("Unknown relation OID: {}", rel_oid))?;
if num_columns != relation.columns.len() {
return Err(format!(
"Column count mismatch: {} vs {}",
num_columns,
relation.columns.len()
));
}
for _ in 0..num_columns {
let marker = Self::read_u8(buf)?;
let value = match marker as char {
'n' => ColumnValue::Null,
'u' => ColumnValue::UnchangedToasted,
't' => {
let len = Self::read_i32(buf)? as usize;
if buf.len() < len {
return Err("Not enough bytes for text column".to_string());
}
let data = buf.split_to(len).to_vec();
let text = String::from_utf8(data)
.map_err(|e| format!("Invalid UTF-8 in column: {}", e))?;
ColumnValue::Text(text)
}
'b' => {
let len = Self::read_i32(buf)? as usize;
if buf.len() < len {
return Err("Not enough bytes for binary column".to_string());
}
let data = buf.split_to(len).to_vec();
ColumnValue::Binary(data)
}
_ => return Err(format!("Unknown column marker: {}", marker as char)),
};
values.push(value);
}
Ok(TupleData { values })
}
/// Get a relation by OID (used for cache invalidation mapping)
pub fn get_relation(&self, oid: u32) -> Option<&RelationInfo> {
self.relations.get(&oid)
}
/// Get all cached relations
pub fn relations(&self) -> &HashMap<u32, RelationInfo> {
&self.relations
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_message_parser_creation() {
let parser = MessageParser::new();
assert!(parser.relations.is_empty());
}
}