diff --git a/Cargo.lock b/Cargo.lock index 9c2a0b455e..f62b1c3ecd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1245,7 +1245,7 @@ dependencies = [ "tar", "thiserror", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-stream", "tokio-util", "tracing", @@ -1351,7 +1351,7 @@ dependencies = [ "storage_broker", "thiserror", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-util", "toml", "toml_edit", @@ -3620,8 +3620,8 @@ dependencies = [ "pageserver_compaction", "pin-project-lite", "postgres", - "postgres-protocol", - "postgres-types", + "postgres-protocol 0.6.4", + "postgres-types 0.2.4", "postgres_backend", "postgres_connection", "postgres_ffi", @@ -3649,7 +3649,7 @@ dependencies = [ "tokio", "tokio-epoll-uring", "tokio-io-timeout", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-stream", "tokio-tar", "tokio-util", @@ -3707,7 +3707,7 @@ dependencies = [ "serde", "thiserror", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-stream", "tokio-util", "utils", @@ -4006,14 +4006,31 @@ dependencies = [ [[package]] name = "postgres" version = "0.19.4" -source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2#20031d7a9ee1addeae6e0968e3899ae6bf01cee2" dependencies = [ "bytes", "fallible-iterator", "futures-util", "log", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", +] + +[[package]] +name = "postgres-protocol" +version = "0.6.4" +dependencies = [ + "base64 0.20.0", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "lazy_static", + "md-5", + "memchr", + "rand 0.8.5", + "sha2", + "stringprep", + "tokio", ] [[package]] @@ -4035,6 +4052,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "postgres-types" +version = "0.2.4" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol 0.6.4", + "serde", + "serde_json", +] + [[package]] name = "postgres-types" version = "0.2.4" @@ -4042,7 +4070,7 @@ source = "git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1 dependencies = [ "bytes", "fallible-iterator", - "postgres-protocol", + "postgres-protocol 0.6.4 (git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2)", "serde", "serde_json", ] @@ -4060,7 +4088,7 @@ dependencies = [ "serde", "thiserror", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-postgres-rustls", "tokio-rustls 0.26.0", "tokio-util", @@ -4075,7 +4103,7 @@ dependencies = [ "itertools 0.10.5", "once_cell", "postgres", - "tokio-postgres", + "tokio-postgres 0.7.7", "url", ] @@ -4127,7 +4155,7 @@ dependencies = [ "byteorder", "bytes", "itertools 0.10.5", - "postgres-protocol", + "postgres-protocol 0.6.4", "rand 0.8.5", "serde", "thiserror", @@ -4313,7 +4341,7 @@ dependencies = [ "parquet_derive", "pbkdf2", "pin-project-lite", - "postgres-protocol", + "postgres-protocol 0.6.4", "postgres_backend", "pq_proto", "prometheus", @@ -4348,7 +4376,7 @@ dependencies = [ "tikv-jemalloc-ctl", "tikv-jemallocator", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-postgres-rustls", "tokio-rustls 0.26.0", "tokio-tungstenite", @@ -5155,9 +5183,10 @@ dependencies = [ "hyper 0.14.30", "metrics", "once_cell", + "pageserver_api", "parking_lot 0.12.1", "postgres", - "postgres-protocol", + "postgres-protocol 0.6.4", "postgres_backend", "postgres_ffi", "pq_proto", @@ -5177,7 +5206,7 @@ dependencies = [ "thiserror", "tokio", "tokio-io-timeout", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-stream", "tokio-tar", "tokio-util", @@ -5185,6 +5214,7 @@ dependencies = [ "tracing-subscriber", "url", "utils", + "wal_decoder", "walproposer", "workspace_hack", ] @@ -5821,7 +5851,7 @@ dependencies = [ "serde_json", "storage_controller_client", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-postgres-rustls", "tokio-stream", "tokio-util", @@ -6218,6 +6248,28 @@ dependencies = [ "syn 2.0.52", ] +[[package]] +name = "tokio-postgres" +version = "0.7.7" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot 0.12.1", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol 0.6.4", + "postgres-types 0.2.4", + "socket2", + "tokio", + "tokio-util", +] + [[package]] name = "tokio-postgres" version = "0.7.7" @@ -6234,8 +6286,8 @@ dependencies = [ "percent-encoding", "phf", "pin-project-lite", - "postgres-protocol", - "postgres-types", + "postgres-protocol 0.6.4 (git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2)", + "postgres-types 0.2.4 (git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2)", "socket2", "tokio", "tokio-util", @@ -6250,7 +6302,7 @@ dependencies = [ "ring", "rustls 0.23.16", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-rustls 0.26.0", "x509-certificate", ] @@ -6833,7 +6885,7 @@ dependencies = [ "serde_json", "sysinfo", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7", "tokio-util", "tracing", "tracing-subscriber", @@ -7340,7 +7392,7 @@ dependencies = [ "num-traits", "once_cell", "parquet", - "postgres-types", + "postgres-types 0.2.4 (git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2)", "prettyplease", "proc-macro2", "prost", @@ -7365,7 +7417,7 @@ dependencies = [ "time", "time-macros", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.7 (git+https://github.com/neondatabase/rust-postgres.git?rev=20031d7a9ee1addeae6e0968e3899ae6bf01cee2)", "tokio-rustls 0.26.0", "tokio-stream", "tokio-util", diff --git a/Cargo.toml b/Cargo.toml index 8207726caa..265e270029 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -214,10 +214,14 @@ log = "0.4" # # When those proxy changes are re-applied (see PR #8747), we can switch using # the tip of the 'neon' branch again. -postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } -postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } -postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } -tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +# postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +# postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +# postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +# tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +postgres = { path = "../../.cargo/git/checkouts/rust-postgres-e2c00088c8e2b112/20031d7/postgres" } +postgres-protocol = { path = "../../.cargo/git/checkouts/rust-postgres-e2c00088c8e2b112/20031d7/postgres-protocol" } +postgres-types = { path = "../../.cargo/git/checkouts/rust-postgres-e2c00088c8e2b112/20031d7/postgres-types" } +tokio-postgres = { path = "../../.cargo/git/checkouts/rust-postgres-e2c00088c8e2b112/20031d7/tokio-postgres" } ## Local libraries compute_api = { version = "0.1", path = "./libs/compute_api/" } @@ -255,7 +259,8 @@ tonic-build = "0.12" [patch.crates-io] # Needed to get `tokio-postgres-rustls` to depend on our fork. -tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } +tokio-postgres = { path = "../../.cargo/git/checkouts/rust-postgres-e2c00088c8e2b112/20031d7/tokio-postgres" } +# tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev = "20031d7a9ee1addeae6e0968e3899ae6bf01cee2" } ################# Binary contents sections diff --git a/libs/pageserver_api/src/key.rs b/libs/pageserver_api/src/key.rs index 4505101ea6..c39460010d 100644 --- a/libs/pageserver_api/src/key.rs +++ b/libs/pageserver_api/src/key.rs @@ -24,7 +24,7 @@ pub struct Key { /// When working with large numbers of Keys in-memory, it is more efficient to handle them as i128 than as /// a struct of fields. -#[derive(Clone, Copy, Hash, PartialEq, Eq, Ord, PartialOrd, Serialize, Deserialize)] +#[derive(Clone, Copy, Hash, PartialEq, Eq, Ord, PartialOrd, Serialize, Deserialize, Debug)] pub struct CompactKey(i128); /// The storage key size. diff --git a/libs/pq_proto/Cargo.toml b/libs/pq_proto/Cargo.toml index 9524a1490d..98b668d9ae 100644 --- a/libs/pq_proto/Cargo.toml +++ b/libs/pq_proto/Cargo.toml @@ -13,3 +13,4 @@ rand.workspace = true tokio = { workspace = true, features = ["io-util"] } thiserror.workspace = true serde.workspace = true +# wal_decoder.workspace = true diff --git a/libs/pq_proto/src/lib.rs b/libs/pq_proto/src/lib.rs index 9ffaaba584..a5933e9427 100644 --- a/libs/pq_proto/src/lib.rs +++ b/libs/pq_proto/src/lib.rs @@ -562,6 +562,7 @@ pub enum BeMessage<'a> { options: &'a [&'a str], }, KeepAlive(WalSndKeepAlive), + InterpretedWalRecord(&'a [u8]), } /// Common shorthands. @@ -996,6 +997,14 @@ impl BeMessage<'_> { Ok(()) })? } + + BeMessage::InterpretedWalRecord(rec) => { + buf.put_u8(b'd'); // arbitrary? + write_body(buf, |buf| { + buf.put_u8(b'0'); + buf.put_slice(rec); + }); + } } Ok(()) } diff --git a/libs/wal_decoder/src/serialized_batch.rs b/libs/wal_decoder/src/serialized_batch.rs index 632603cc8b..0426711c7a 100644 --- a/libs/wal_decoder/src/serialized_batch.rs +++ b/libs/wal_decoder/src/serialized_batch.rs @@ -30,7 +30,7 @@ static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; BLCKSZ as usize]); /// relation sizes. In the case of "observed" values, we only need to know /// the key and LSN, so two types of metadata are supported to save on network /// bandwidth. -#[derive(Serialize, Deserialize)] +#[derive(Serialize, Deserialize, Debug)] pub enum ValueMeta { Serialized(SerializedValueMeta), Observed(ObservedValueMeta), @@ -77,7 +77,7 @@ impl PartialEq for OrderedValueMeta { impl Eq for OrderedValueMeta {} /// Metadata for a [`Value`] serialized into the batch. -#[derive(Serialize, Deserialize)] +#[derive(Serialize, Deserialize, Debug)] pub struct SerializedValueMeta { pub key: CompactKey, pub lsn: Lsn, @@ -89,7 +89,7 @@ pub struct SerializedValueMeta { } /// Metadata for a [`Value`] observed by the batch -#[derive(Serialize, Deserialize)] +#[derive(Serialize, Deserialize, Debug)] pub struct ObservedValueMeta { pub key: CompactKey, pub lsn: Lsn, diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs index 7b106569a4..c8ae320b5f 100644 --- a/pageserver/src/pgdatadir_mapping.rs +++ b/pageserver/src/pgdatadir_mapping.rs @@ -1164,12 +1164,19 @@ impl<'a> DatadirModification<'a> { .get_rel_exists(rel, Version::Modified(self), ctx) .await? { + tracing::info!("Creating relation {rel:?} at lsn {}", self.get_lsn()); + // create it with 0 size initially, the logic below will extend it self.put_rel_creation(rel, 0, ctx) .await .context("Relation Error")?; Ok(0) } else { + tracing::info!( + "Skipping relation {rel:?} creation at lsn {}", + self.get_lsn() + ); + self.tline .get_rel_size(rel, Version::Modified(self), ctx) .await @@ -1210,6 +1217,8 @@ impl<'a> DatadirModification<'a> { shard: &ShardIdentity, ctx: &RequestContext, ) -> anyhow::Result<()> { + tracing::info!("Ingesting batch with metadata: {:?}", batch.metadata); + let mut gaps_at_lsns = Vec::default(); for meta in batch.metadata.iter() { diff --git a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs index 34bf959058..11dc4e9a12 100644 --- a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs +++ b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs @@ -36,7 +36,7 @@ use crate::{ use postgres_backend::is_expected_io_error; use postgres_connection::PgConnectionConfig; use postgres_ffi::waldecoder::WalStreamDecoder; -use utils::{id::NodeId, lsn::Lsn}; +use utils::{bin_ser::BeSer, id::NodeId, lsn::Lsn}; use utils::{pageserver_feedback::PageserverFeedback, sync::gate::GateError}; /// Status of the connection. @@ -278,6 +278,7 @@ pub(super) async fn handle_walreceiver_connection( // fails (e.g. in walingest), we still want to know latests LSNs from the safekeeper. match &replication_message { ReplicationMessage::XLogData(xlog_data) => { + // TODO(vlad) Is this crap needed? connection_status.latest_connection_update = now; connection_status.commit_lsn = Some(Lsn::from(xlog_data.wal_end())); connection_status.streaming_lsn = Some(Lsn::from( @@ -299,6 +300,20 @@ pub(super) async fn handle_walreceiver_connection( } let status_update = match replication_message { + ReplicationMessage::RawInterpretedWalRecord(raw_interpreted) => { + let interpreted = InterpretedWalRecord::des(&raw_interpreted).unwrap(); + let end_lsn = interpreted.end_lsn; + + let mut modification = timeline.begin_modification(end_lsn); + walingest + .ingest_record(interpreted, &mut modification, &ctx) + .await + .with_context(|| format!("could not ingest record at {}", end_lsn))?; + modification.commit(&ctx).await?; + + Some(end_lsn) + } + ReplicationMessage::XLogData(xlog_data) => { // Pass the WAL data to the decoder, and see if we can decode // more records as a result. diff --git a/proxy/src/compute.rs b/proxy/src/compute.rs index ca4a348ed8..77071abd0e 100644 --- a/proxy/src/compute.rs +++ b/proxy/src/compute.rs @@ -316,6 +316,7 @@ impl ConnCfg { let client_config = client_config.with_no_client_auth(); let mut mk_tls = tokio_postgres_rustls::MakeRustlsConnect::new(client_config); + // TODO(vlad): que? let tls = >::make_tls_connect( &mut mk_tls, host, diff --git a/safekeeper/Cargo.toml b/safekeeper/Cargo.toml index 0372418598..82ced2b86c 100644 --- a/safekeeper/Cargo.toml +++ b/safekeeper/Cargo.toml @@ -58,6 +58,7 @@ sd-notify.workspace = true storage_broker.workspace = true tokio-stream.workspace = true utils.workspace = true +wal_decoder.workspace = true workspace_hack.workspace = true diff --git a/safekeeper/src/send_wal.rs b/safekeeper/src/send_wal.rs index 6d94ff98b1..9b3c5bd779 100644 --- a/safekeeper/src/send_wal.rs +++ b/safekeeper/src/send_wal.rs @@ -11,10 +11,12 @@ use crate::wal_storage::WalReader; use crate::GlobalTimelines; use anyhow::{bail, Context as AnyhowContext}; use bytes::Bytes; +use pageserver_api::shard::ShardIdentity; use parking_lot::Mutex; use postgres_backend::PostgresBackend; use postgres_backend::{CopyStreamHandlerEnd, PostgresBackendReader, QueryError}; use postgres_ffi::get_current_timestamp; +use postgres_ffi::waldecoder::WalStreamDecoder; use postgres_ffi::{TimestampTz, MAX_SEND_SIZE}; use pq_proto::{BeMessage, WalSndKeepAlive, XLogDataBody}; use serde::{Deserialize, Serialize}; @@ -22,6 +24,8 @@ use tokio::io::{AsyncRead, AsyncWrite}; use utils::failpoint_support; use utils::id::TenantTimelineId; use utils::pageserver_feedback::PageserverFeedback; +use utils::postgres_client::{PAGESERVER_SAFEKEEPER_PROTO_VERSION, POSTGRES_PROTO_VERSION}; +use wal_decoder::models::InterpretedWalRecord; use std::cmp::{max, min}; use std::net::SocketAddr; @@ -377,6 +381,10 @@ impl Drop for WalSenderGuard { } impl SafekeeperPostgresHandler { + pub fn protocol_version(&self) -> u8 { + self.protocol_version.unwrap_or(POSTGRES_PROTO_VERSION) + } + /// Wrapper around handle_start_replication_guts handling result. Error is /// handled here while we're still in walsender ttid span; with API /// extension, this can probably be moved into postgres_backend. @@ -412,6 +420,7 @@ impl SafekeeperPostgresHandler { let appname = self.appname.clone(); // Use a guard object to remove our entry from the timeline when we are done. + // TODO(vlad): maybe thread shard stuff into here let ws_guard = Arc::new(tli.get_walsenders().register( self.ttid, *pgb.get_peer_addr(), @@ -475,9 +484,10 @@ impl SafekeeperPostgresHandler { tli, }; + let protocol_version = self.protocol_version(); let res = tokio::select! { // todo: add read|write .context to these errors - r = sender.run() => r, + r = sender.run(protocol_version, self.shard.as_ref()) => r, r = reply_reader.run() => r, }; @@ -560,7 +570,120 @@ impl WalSender<'_, IO> { /// /// Err(CopyStreamHandlerEnd) is always returned; Result is used only for ? /// convenience. - async fn run(&mut self) -> Result<(), CopyStreamHandlerEnd> { + /// TODO(vlad): add a run variant which accumulates a full wall record + /// and interprets it. + async fn run( + &mut self, + protocol_version: u8, + shard: Option<&ShardIdentity>, + ) -> Result<(), CopyStreamHandlerEnd> { + match protocol_version { + POSTGRES_PROTO_VERSION => self.run_wal_sender().await, + PAGESERVER_SAFEKEEPER_PROTO_VERSION => { + self.run_interpreted_record_sender(shard.unwrap()).await + } + // TODO: make the proto version an enum + _ => unreachable!(), + } + } + + async fn run_interpreted_record_sender( + &mut self, + shard: &ShardIdentity, + ) -> Result<(), CopyStreamHandlerEnd> { + let pg_version = self.tli.tli.get_state().await.1.server.pg_version / 10000; + let mut wal_decoder = WalStreamDecoder::new(self.start_pos, pg_version); + + loop { + // Wait for the next portion if it is not there yet, or just + // update our end of WAL available for sending value, we + // communicate it to the receiver. + self.wait_wal().await?; + assert!( + self.end_pos > self.start_pos, + "nothing to send after waiting for WAL" + ); + + // try to send as much as available, capped by MAX_SEND_SIZE + let mut chunk_end_pos = self.start_pos + MAX_SEND_SIZE as u64; + // if we went behind available WAL, back off + if chunk_end_pos >= self.end_pos { + chunk_end_pos = self.end_pos; + } else { + // If sending not up to end pos, round down to page boundary to + // avoid breaking WAL record not at page boundary, as protocol + // demands. See walsender.c (XLogSendPhysical). + chunk_end_pos = chunk_end_pos + .checked_sub(chunk_end_pos.block_offset()) + .unwrap(); + } + let send_size = (chunk_end_pos.0 - self.start_pos.0) as usize; + let send_buf = &mut self.send_buf[..send_size]; + let send_size: usize; + { + // If uncommitted part is being pulled, check that the term is + // still the expected one. + let _term_guard = if let Some(t) = self.term { + Some(self.tli.acquire_term(t).await?) + } else { + None + }; + // Read WAL into buffer. send_size can be additionally capped to + // segment boundary here. + send_size = self.wal_reader.read(send_buf).await? + }; + let send_buf = &send_buf[..send_size]; + + wal_decoder.feed_bytes(send_buf); + + // TODO(vlad): implement error handling here + while let Some((record_end_lsn, recdata)) = wal_decoder.poll_decode().unwrap() { + assert!(record_end_lsn.is_aligned()); + + // Deserialize and interpret WAL record + let interpreted = InterpretedWalRecord::from_bytes_filtered( + recdata, + shard, + record_end_lsn, + pg_version, + ) + .unwrap(); + + let mut buf = Vec::new(); + interpreted.ser_into(&mut buf).unwrap(); + + self.pgb + .write_message(&BeMessage::InterpretedWalRecord(buf.as_slice())) + .await?; + } + + // and send it + // self.pgb + // .write_message(&BeMessage::XLogData(XLogDataBody { + // wal_start: self.start_pos.0, + // wal_end: self.end_pos.0, + // timestamp: get_current_timestamp(), + // data: send_buf, + // })) + // .await?; + + // if let Some(appname) = &self.appname { + // if appname == "replica" { + // failpoint_support::sleep_millis_async!("sk-send-wal-replica-sleep"); + // } + // } + // trace!( + // "sent {} bytes of WAL {}-{}", + // send_size, + // self.start_pos, + // self.start_pos + send_size as u64 + // ); + + self.start_pos += send_size as u64; + } + } + + async fn run_wal_sender(&mut self) -> Result<(), CopyStreamHandlerEnd> { loop { // Wait for the next portion if it is not there yet, or just // update our end of WAL available for sending value, we