rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-08-13 01:33:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix restoring non-relational data during compute node startup

Browse Source
This commit is contained in:
Konstantin Knizhnik
2021-05-20 14:05:43 +03:00
parent 06f96f9600
commit 20b6279beb
11 changed files with 44 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
control_plane/src/compute.rs
-2
View File
@@ -256,7 +256,6 @@ impl PostgresNode {
// new data directory
pub fn init_from_page_server(&self) -> Result<()> {
let pgdata = self.pgdata();
println!(
"Extracting base backup to create postgres instance: path={} port={}",
pgdata.display(),
@@ -348,7 +347,6 @@ impl PostgresNode {
fs::create_dir_all(self.pgdata().join("pg_wal"))?;
fs::create_dir_all(self.pgdata().join("pg_wal").join("archive_status"))?;
self.pg_resetwal(&["-f"])?;
Ok(())
}
integration_tests/tests/test_wal_acceptor.rs
+1 -2
View File
@@ -3,7 +3,7 @@ use std::sync::Arc;
use std::time::SystemTime;
use std::{thread, time};
use control_plane::compute::ComputeControlPlane;
use control_plane::compute::{PostgresNode, ComputeControlPlane};
use integration_tests;
use integration_tests::PostgresNodeExt;
@@ -26,7 +26,6 @@ fn start_node_with_wal_proposer(
}
#[test]
//#[ignore]
fn test_embedded_wal_proposer() {
let local_env = integration_tests::create_test_env("test_embedded_wal_proposer");
pageserver/src/basebackup.rs
+1
View File
@@ -149,6 +149,7 @@ fn add_pgcontrol_file(
let mut checkpoint = postgres_ffi::decode_checkpoint(checkpoint_bytes)?;
checkpoint.redo = lsn.0;
checkpoint.nextXid.value += 1;
// TODO: When we restart master there are no active transaction and oldestXid is
// equal to nextXid if there are no prepared transactions.
// Let's ignore them for a while...
pageserver/src/branches.rs
+6 -34
View File
@@ -8,7 +8,7 @@ use anyhow::{anyhow, bail, Context, Result};
use bytes::Bytes;
use fs::File;
use fs_extra;
use postgres_ffi::xlog_utils;
use postgres_ffi::{pg_constants, xlog_utils};
use rand::Rng;
use serde::{Deserialize, Serialize};
use std::env;
@@ -99,9 +99,6 @@ pub fn init_repo(conf: &PageServerConf, repo_dir: &Path) -> Result<()> {
// Remove pg_wal
fs::remove_dir_all(tmppath.join("pg_wal"))?;
force_crash_recovery(&tmppath)?;
println!("updated pg_control");
let target = timelinedir.join("snapshots").join(&lsnstr);
fs::rename(tmppath, &target)?;
@@ -229,7 +226,7 @@ pub(crate) fn create_branch(
&oldtimelinedir.join("wal"),
&newtimelinedir.join("wal"),
startpoint.lsn,
16 * 1024 * 1024, // FIXME: assume default WAL segment size
pg_constants::WAL_SEGMENT_SIZE,
)?;
Ok(BranchInfo {
@@ -310,31 +307,6 @@ fn parse_point_in_time(conf: &PageServerConf, s: &str) -> Result<PointInTime> {
bail!("could not parse point-in-time {}", s);
}
// If control file says the cluster was shut down cleanly, modify it, to mark
// it as crashed. That forces crash recovery when you start the cluster.
//
// FIXME:
// We currently do this to the initial snapshot in "zenith init". It would
// be more natural to do this when the snapshot is restored instead, but we
// currently don't have any code to create new snapshots, so it doesn't matter
// Or better yet, use a less hacky way of putting the cluster into recovery.
// Perhaps create a backup label file in the data directory when it's restored.
fn force_crash_recovery(datadir: &Path) -> Result<()> {
// Read in the control file
let controlfilepath = datadir.to_path_buf().join("global").join("pg_control");
let mut controlfile =
postgres_ffi::decode_pg_control(Bytes::from(fs::read(controlfilepath.as_path())?))?;
controlfile.state = postgres_ffi::DBState_DB_IN_PRODUCTION;
fs::write(
controlfilepath.as_path(),
postgres_ffi::encode_pg_control(controlfile),
)?;
Ok(())
}
fn create_timeline(conf: &PageServerConf, ancestor: Option<PointInTime>) -> Result<ZTimelineId> {
// Create initial timeline
let mut tli_buf = [0u8; 16];
@@ -361,7 +333,7 @@ fn create_timeline(conf: &PageServerConf, ancestor: Option<PointInTime>) -> Resu
/// If the given LSN is in the middle of a segment, the last segment containing it
/// is written out as .partial, and padded with zeros.
///
fn copy_wal(src_dir: &Path, dst_dir: &Path, upto: Lsn, wal_seg_size: u64) -> Result<()> {
fn copy_wal(src_dir: &Path, dst_dir: &Path, upto: Lsn, wal_seg_size: usize) -> Result<()> {
let last_segno = upto.segment_number(wal_seg_size);
let last_segoff = upto.segment_offset(wal_seg_size);
@@ -391,11 +363,11 @@ fn copy_wal(src_dir: &Path, dst_dir: &Path, upto: Lsn, wal_seg_size: u64) -> Res
let src_file = File::open(entry.path())?;
let mut dst_file = File::create(dst_dir.join(&dst_fname))?;
std::io::copy(&mut src_file.take(copylen), &mut dst_file)?;
std::io::copy(&mut src_file.take(copylen as u64), &mut dst_file)?;
if copylen < wal_seg_size {
std::io::copy(
&mut std::io::repeat(0).take(wal_seg_size - copylen),
&mut std::io::repeat(0).take((wal_seg_size - copylen) as u64),
&mut dst_file,
)?;
}
@@ -407,7 +379,7 @@ fn copy_wal(src_dir: &Path, dst_dir: &Path, upto: Lsn, wal_seg_size: u64) -> Res
// Find the end of valid WAL in a wal directory
pub fn find_end_of_wal(conf: &PageServerConf, timeline: ZTimelineId) -> Result<Lsn> {
let waldir = conf.timeline_path(timeline).join("wal");
let (lsn, _tli) = xlog_utils::find_end_of_wal(&waldir, 16 * 1024 * 1024, true);
let (lsn, _tli) = xlog_utils::find_end_of_wal(&waldir, pg_constants::WAL_SEGMENT_SIZE, true);
Ok(Lsn(lsn))
}
pageserver/src/page_service.rs
+1 -1
View File
@@ -925,7 +925,7 @@ impl Connection {
// find latest snapshot
let snapshot_lsn =
restore_local_repo::find_latest_snapshot(&self.conf, timelineid).unwrap();
let req_lsn = lsn.unwrap_or(snapshot_lsn);
let req_lsn = lsn.unwrap_or(timeline.get_last_valid_lsn());
basebackup::send_tarball_at_lsn(
&mut CopyDataSink { stream },
timelineid,
pageserver/src/repository/rocksdb.rs
+4 -8
View File
@@ -707,11 +707,11 @@ impl Timeline for RocksTimeline {
/// Get range [begin,end) of stored blocks. Used mostly for SMGR pseudorelations
/// but can be also applied to normal relations.
fn get_range(&self, rel: RelTag, lsn: Lsn) -> Result<(u32, u32)> {
let lsn = self.wait_lsn(lsn)?;
let _lsn = self.wait_lsn(lsn)?;
let mut key = CacheKey {
// minimal key to start with
tag: BufferTag { rel, blknum: 0 },
lsn,
lsn: Lsn(0),
};
let mut iter = self.db.raw_iterator();
iter.seek(key.to_bytes()); // locate first entry
@@ -817,12 +817,8 @@ impl Timeline for RocksTimeline {
///
fn get_page_image(&self, tag: BufferTag, lsn: Lsn) -> Result<Option<Bytes>> {
let key = CacheKey { tag, lsn };
let mut key_buf = BytesMut::new();
key.pack(&mut key_buf);
if let Some(bytes) = self.db.get(&key_buf[..])? {
let mut buf = BytesMut::new();
buf.extend_from_slice(&bytes);
let content = CacheEntryContent::unpack(&mut buf);
if let Some(bytes) = self.db.get(key.to_bytes())? {
let content = CacheEntryContent::from_slice(&bytes);
if let CacheEntryContent::PageImage(img) = content {
return Ok(Some(img));
}
pageserver/src/restore_local_repo.rs
+6 -1
View File
@@ -454,6 +454,8 @@ fn restore_wal(timeline: &dyn Timeline, timelineid: ZTimelineId, startpoint: Lsn
if rec.is_err() {
// Assume that an error means we've reached the end of
// a partial WAL record. So that's ok.
trace!("WAL decoder error {:?}", rec);
waldecoder.set_position(Lsn((segno + 1) * pg_constants::WAL_SEGMENT_SIZE as u64));
break;
}
if let Some((lsn, recdata)) = rec.unwrap() {
@@ -466,7 +468,10 @@ fn restore_wal(timeline: &dyn Timeline, timelineid: ZTimelineId, startpoint: Lsn
nrecords += 1;
}
info!("restored {} records from WAL file {}", nrecords, filename);
info!(
"restored {} records from WAL file {} at {}",
nrecords, filename, last_lsn
);
segno += 1;
offset = 0;
pageserver/src/waldecoder.rs
+18 -3
View File
@@ -84,6 +84,13 @@ impl WalStreamDecoder {
}
}
pub fn set_position(&mut self, lsn: Lsn) {
self.lsn = lsn;
self.contlen = 0;
self.padlen = 0;
self.inputbuf.clear();
}
pub fn feed_bytes(&mut self, buf: &[u8]) {
self.inputbuf.extend_from_slice(buf);
}
@@ -957,7 +964,7 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
trace!("XLOG_TBLSPC_DROP is not handled yet");
}
} else if xlogrec.xl_rmid == pg_constants::RM_HEAP_ID {
let info = xlogrec.xl_info & pg_constants::XLOG_XACT_OPMASK;
let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
let blkno = blocks[0].blkno / pg_constants::HEAPBLOCKS_PER_PAGE as u32;
if info == pg_constants::XLOG_HEAP_INSERT {
let xlrec = XlHeapInsert::decode(&mut buf);
@@ -1011,7 +1018,7 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
}
}
} else if xlogrec.xl_rmid == pg_constants::RM_HEAP2_ID {
let info = xlogrec.xl_info & pg_constants::XLOG_XACT_OPMASK;
let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
let xlrec = XlHeapMultiInsert::decode(&mut buf);
if (xlrec.flags
@@ -1030,7 +1037,7 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
}
}
} else if xlogrec.xl_rmid == pg_constants::RM_MULTIXACT_ID {
let info = xlogrec.xl_info & pg_constants::XLOG_XACT_OPMASK;
let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
if info == pg_constants::XLOG_MULTIXACT_ZERO_OFF_PAGE {
let mut blk = DecodedBkpBlock::new();
blk.forknum = pg_constants::PG_MXACT_OFFSETS_FORKNUM;
@@ -1117,6 +1124,14 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
blk.rnode_relnode = 0;
blk.will_init = true;
blocks.push(blk);
} else if xlogrec.xl_rmid == pg_constants::RM_XLOG_ID {
let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
if info == pg_constants::XLOG_NEXTOID {
let next_oid = buf.get_u32_le();
if next_oid > checkpoint.nextOid {
checkpoint.nextOid = next_oid;
}
}
}
DecodedWALRecord {
pageserver/src/walreceiver.rs
+1
View File
@@ -196,6 +196,7 @@ fn walreceiver_main(
while let Some((lsn, recdata)) = waldecoder.poll_decode()? {
let old_checkpoint_bytes = encode_checkpoint(checkpoint);
//info!("Decode WAL record at LSN {}", lsn);
let decoded = decode_wal_record(&mut checkpoint, recdata.clone());
timeline.save_decoded_record(decoded, recdata, lsn)?;
pageserver/src/walredo.rs
+5 -1
View File
@@ -293,6 +293,10 @@ impl PostgresRedoManagerInternal {
let mut status = 0;
if info == pg_constants::XLOG_XACT_COMMIT {
status = pg_constants::TRANSACTION_STATUS_COMMITTED;
info!(
"Mark transaction {} as committed at LSN {}",
xlogrec.xl_xid, lsn
);
transaction_id_set_status(xlogrec.xl_xid, status, &mut page);
//handle subtrans
let _xact_time = buf.get_i64_le();
@@ -350,7 +354,7 @@ impl PostgresRedoManagerInternal {
record.main_data_offset, record.rec.len());
}
} else if xlogrec.xl_rmid == pg_constants::RM_MULTIXACT_ID {
let info = xlogrec.xl_info & pg_constants::XLOG_XACT_OPMASK;
let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
if info == pg_constants::XLOG_MULTIXACT_ZERO_OFF_PAGE {
page.copy_from_slice(&ZERO_PAGE);
} else if info == pg_constants::XLOG_MULTIXACT_ZERO_MEM_PAGE {
postgres_ffi/src/pg_constants.rs
+1
View File
@@ -83,6 +83,7 @@ pub const XACT_XINFO_HAS_TWOPHASE: u32 = 1u32 << 4;
// pub const XACT_XINFO_HAS_GID: u32 = 1u32 << 7;
// From pg_control.h and rmgrlist.h
pub const XLOG_NEXTOID: u8 = 0x30;
pub const XLOG_SWITCH: u8 = 0x40;
pub const XLOG_SMGR_TRUNCATE: u8 = 0x20;