Line data Source code
1 : //! Tests for postgres_ffi xlog_utils module. Put it here to break cyclic dependency.
2 :
3 : use super::*;
4 : use crate::{error, info};
5 : use regex::Regex;
6 : use std::cmp::min;
7 : use std::fs::{self, File};
8 : use std::io::Write;
9 : use std::{env, str::FromStr};
10 : use utils::const_assert;
11 : use utils::lsn::Lsn;
12 :
13 54 : fn init_logging() {
14 54 : let _ = env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(format!(
15 54 : "crate=info,postgres_ffi::{PG_MAJORVERSION}::xlog_utils=trace"
16 54 : )))
17 54 : .is_test(true)
18 54 : .try_init();
19 54 : }
20 :
21 : /// Test that find_end_of_wal returns the same results as pg_dump on various
22 : /// WALs created by Crafter.
23 54 : fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
24 54 : use crate::*;
25 54 :
26 54 : let pg_version = PG_MAJORVERSION[1..3].parse::<u32>().unwrap();
27 54 :
28 54 : // Craft some WAL
29 54 : let top_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
30 54 : .join("..")
31 54 : .join("..")
32 54 : .join("..");
33 54 : let cfg = Conf {
34 54 : pg_version,
35 54 : pg_distrib_dir: top_path.join("pg_install"),
36 54 : datadir: top_path.join(format!("test_output/{}-{PG_MAJORVERSION}", test_name)),
37 54 : };
38 54 : if cfg.datadir.exists() {
39 45 : fs::remove_dir_all(&cfg.datadir).unwrap();
40 45 : }
41 54 : cfg.initdb().unwrap();
42 54 : let srv = cfg.start_server().unwrap();
43 54 : let intermediate_lsns = C::craft(&mut srv.connect_with_timeout().unwrap()).unwrap();
44 54 : let intermediate_lsns: Vec<Lsn> = intermediate_lsns
45 54 : .iter()
46 90 : .map(|&lsn| u64::from(lsn).into())
47 54 : .collect();
48 54 : // Kill postgres. Note that it might have inserted to WAL something after
49 54 : // 'craft' did its job.
50 54 : srv.kill();
51 54 :
52 54 : // Check find_end_of_wal on the initial WAL
53 54 : let last_segment = cfg
54 54 : .wal_dir()
55 54 : .read_dir()
56 54 : .unwrap()
57 144 : .map(|f| f.unwrap().file_name().into_string().unwrap())
58 144 : .filter(|fname| IsXLogFileName(fname))
59 54 : .max()
60 54 : .unwrap();
61 54 : let expected_end_of_wal = find_pg_waldump_end_of_wal(&cfg, &last_segment);
62 144 : for start_lsn in intermediate_lsns
63 54 : .iter()
64 54 : .chain(std::iter::once(&expected_end_of_wal))
65 : {
66 : // Erase all WAL before `start_lsn` to ensure it's not used by `find_end_of_wal`.
67 : // We assume that `start_lsn` is non-decreasing.
68 144 : info!(
69 144 : "Checking with start_lsn={}, erasing WAL before it",
70 : start_lsn
71 : );
72 396 : for file in fs::read_dir(cfg.wal_dir()).unwrap().flatten() {
73 396 : let fname = file.file_name().into_string().unwrap();
74 396 : if !IsXLogFileName(&fname) {
75 144 : continue;
76 252 : }
77 252 : let (segno, _) = XLogFromFileName(&fname, WAL_SEGMENT_SIZE);
78 252 : let seg_start_lsn = XLogSegNoOffsetToRecPtr(segno, 0, WAL_SEGMENT_SIZE);
79 252 : if seg_start_lsn > u64::from(*start_lsn) {
80 36 : continue;
81 216 : }
82 216 : let mut f = File::options().write(true).open(file.path()).unwrap();
83 216 : const ZEROS: [u8; WAL_SEGMENT_SIZE] = [0u8; WAL_SEGMENT_SIZE];
84 216 : f.write_all(
85 216 : &ZEROS[0..min(
86 216 : WAL_SEGMENT_SIZE,
87 216 : (u64::from(*start_lsn) - seg_start_lsn) as usize,
88 216 : )],
89 216 : )
90 216 : .unwrap();
91 : }
92 144 : check_end_of_wal(&cfg, &last_segment, *start_lsn, expected_end_of_wal);
93 : }
94 54 : }
95 :
96 54 : fn find_pg_waldump_end_of_wal(cfg: &crate::Conf, last_segment: &str) -> Lsn {
97 54 : // Get the actual end of WAL by pg_waldump
98 54 : let waldump_output = cfg
99 54 : .pg_waldump("000000010000000000000001", last_segment)
100 54 : .unwrap()
101 54 : .stderr;
102 54 : let waldump_output = std::str::from_utf8(&waldump_output).unwrap();
103 54 : let caps = match Regex::new(r"invalid record length at (.+):")
104 54 : .unwrap()
105 54 : .captures(waldump_output)
106 : {
107 54 : Some(caps) => caps,
108 : None => {
109 0 : error!("Unable to parse pg_waldump's stderr:\n{}", waldump_output);
110 0 : panic!();
111 : }
112 : };
113 54 : let waldump_wal_end = Lsn::from_str(caps.get(1).unwrap().as_str()).unwrap();
114 54 : info!("waldump erred on {}", waldump_wal_end);
115 54 : waldump_wal_end
116 54 : }
117 :
118 144 : fn check_end_of_wal(
119 144 : cfg: &crate::Conf,
120 144 : last_segment: &str,
121 144 : start_lsn: Lsn,
122 144 : expected_end_of_wal: Lsn,
123 144 : ) {
124 144 : // Check end_of_wal on non-partial WAL segment (we treat it as fully populated)
125 144 : // let wal_end = find_end_of_wal(&cfg.wal_dir(), WAL_SEGMENT_SIZE, start_lsn).unwrap();
126 144 : // info!(
127 144 : // "find_end_of_wal returned wal_end={} with non-partial WAL segment",
128 144 : // wal_end
129 144 : // );
130 144 : // assert_eq!(wal_end, expected_end_of_wal_non_partial);
131 144 :
132 144 : // Rename file to partial to actually find last valid lsn, then rename it back.
133 144 : fs::rename(
134 144 : cfg.wal_dir().join(last_segment),
135 144 : cfg.wal_dir().join(format!("{}.partial", last_segment)),
136 144 : )
137 144 : .unwrap();
138 144 : let wal_end = find_end_of_wal(&cfg.wal_dir(), WAL_SEGMENT_SIZE, start_lsn).unwrap();
139 144 : info!(
140 144 : "find_end_of_wal returned wal_end={} with partial WAL segment",
141 : wal_end
142 : );
143 144 : assert_eq!(wal_end, expected_end_of_wal);
144 144 : fs::rename(
145 144 : cfg.wal_dir().join(format!("{}.partial", last_segment)),
146 144 : cfg.wal_dir().join(last_segment),
147 144 : )
148 144 : .unwrap();
149 144 : }
150 :
151 : const_assert!(WAL_SEGMENT_SIZE == 16 * 1024 * 1024);
152 :
153 : #[test]
154 18 : pub fn test_find_end_of_wal_simple() {
155 18 : init_logging();
156 18 : test_end_of_wal::<crate::Simple>("test_find_end_of_wal_simple");
157 18 : }
158 :
159 : #[test]
160 18 : pub fn test_find_end_of_wal_crossing_segment_followed_by_small_one() {
161 18 : init_logging();
162 18 : test_end_of_wal::<crate::WalRecordCrossingSegmentFollowedBySmallOne>(
163 18 : "test_find_end_of_wal_crossing_segment_followed_by_small_one",
164 18 : );
165 18 : }
166 :
167 : #[test]
168 18 : pub fn test_find_end_of_wal_last_crossing_segment() {
169 18 : init_logging();
170 18 : test_end_of_wal::<crate::LastWalRecordCrossingSegment>(
171 18 : "test_find_end_of_wal_last_crossing_segment",
172 18 : );
173 18 : }
174 :
175 : /// Check the math in update_next_xid
176 : ///
177 : /// NOTE: These checks are sensitive to the value of XID_CHECKPOINT_INTERVAL,
178 : /// currently 1024.
179 : #[test]
180 18 : pub fn test_update_next_xid() {
181 18 : let checkpoint_buf = [0u8; size_of::<CheckPoint>()];
182 18 : let mut checkpoint = CheckPoint::decode(&checkpoint_buf).unwrap();
183 18 :
184 18 : checkpoint.nextXid = FullTransactionId { value: 10 };
185 18 : assert_eq!(checkpoint.nextXid.value, 10);
186 :
187 : // The input XID gets rounded up to the next XID_CHECKPOINT_INTERVAL
188 : // boundary
189 18 : checkpoint.update_next_xid(100);
190 18 : assert_eq!(checkpoint.nextXid.value, 1024);
191 :
192 : // No change
193 18 : checkpoint.update_next_xid(500);
194 18 : assert_eq!(checkpoint.nextXid.value, 1024);
195 18 : checkpoint.update_next_xid(1023);
196 18 : assert_eq!(checkpoint.nextXid.value, 1024);
197 :
198 : // The function returns the *next* XID, given the highest XID seen so
199 : // far. So when we pass 1024, the nextXid gets bumped up to the next
200 : // XID_CHECKPOINT_INTERVAL boundary.
201 18 : checkpoint.update_next_xid(1024);
202 18 : assert_eq!(checkpoint.nextXid.value, 2048);
203 18 : }
204 :
205 : #[test]
206 18 : pub fn test_update_next_multixid() {
207 18 : let checkpoint_buf = [0u8; size_of::<CheckPoint>()];
208 18 : let mut checkpoint = CheckPoint::decode(&checkpoint_buf).unwrap();
209 18 :
210 18 : // simple case
211 18 : checkpoint.nextMulti = 20;
212 18 : checkpoint.nextMultiOffset = 20;
213 18 : checkpoint.update_next_multixid(1000, 2000);
214 18 : assert_eq!(checkpoint.nextMulti, 1000);
215 18 : assert_eq!(checkpoint.nextMultiOffset, 2000);
216 :
217 : // No change
218 18 : checkpoint.update_next_multixid(500, 900);
219 18 : assert_eq!(checkpoint.nextMulti, 1000);
220 18 : assert_eq!(checkpoint.nextMultiOffset, 2000);
221 :
222 : // Close to wraparound, but not wrapped around yet
223 18 : checkpoint.nextMulti = 0xffff0000;
224 18 : checkpoint.nextMultiOffset = 0xfffe0000;
225 18 : checkpoint.update_next_multixid(0xffff00ff, 0xfffe00ff);
226 18 : assert_eq!(checkpoint.nextMulti, 0xffff00ff);
227 18 : assert_eq!(checkpoint.nextMultiOffset, 0xfffe00ff);
228 :
229 : // Wraparound
230 18 : checkpoint.update_next_multixid(1, 900);
231 18 : assert_eq!(checkpoint.nextMulti, 1);
232 18 : assert_eq!(checkpoint.nextMultiOffset, 900);
233 :
234 : // Wraparound nextMulti to 0.
235 : //
236 : // It's a bit surprising that nextMulti can be 0, because that's a special value
237 : // (InvalidMultiXactId). However, that's how Postgres does it at multi-xid wraparound:
238 : // nextMulti wraps around to 0, but then when the next multi-xid is assigned, it skips
239 : // the 0 and the next multi-xid actually assigned is 1.
240 18 : checkpoint.nextMulti = 0xffff0000;
241 18 : checkpoint.nextMultiOffset = 0xfffe0000;
242 18 : checkpoint.update_next_multixid(0, 0xfffe00ff);
243 18 : assert_eq!(checkpoint.nextMulti, 0);
244 18 : assert_eq!(checkpoint.nextMultiOffset, 0xfffe00ff);
245 :
246 : // Wraparound nextMultiOffset to 0
247 18 : checkpoint.update_next_multixid(0, 0);
248 18 : assert_eq!(checkpoint.nextMulti, 0);
249 18 : assert_eq!(checkpoint.nextMultiOffset, 0);
250 18 : }
251 :
252 : #[test]
253 18 : pub fn test_encode_logical_message() {
254 18 : let expected = [
255 18 : 64, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 21, 0, 0, 170, 34, 166, 227, 255, 38,
256 18 : 0, 0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 0, 0, 0, 0, 112, 114, 101, 102,
257 18 : 105, 120, 0, 109, 101, 115, 115, 97, 103, 101,
258 18 : ];
259 18 : let actual = encode_logical_message("prefix", "message");
260 18 : assert_eq!(expected, actual[..]);
261 18 : }
|
