Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * archive_waldump.c
4 : : * A generic facility for reading WAL data from tar archives via archive
5 : : * streamer.
6 : : *
7 : : * Portions Copyright (c) 2026, PostgreSQL Global Development Group
8 : : *
9 : : * IDENTIFICATION
10 : : * src/bin/pg_waldump/archive_waldump.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres_fe.h"
16 : :
17 : : #include <unistd.h>
18 : :
19 : : #include "access/xlog_internal.h"
20 : : #include "common/file_perm.h"
21 : : #include "common/hashfn.h"
22 : : #include "common/logging.h"
23 : : #include "fe_utils/simple_list.h"
24 : : #include "pg_waldump.h"
25 : :
26 : : /*
27 : : * How many bytes should we try to read from a file at once?
28 : : */
29 : : #define READ_CHUNK_SIZE (128 * 1024)
30 : :
31 : : /* Temporary directory for spilled WAL segment files */
32 : : char *TmpWalSegDir = NULL;
33 : :
34 : : /*
35 : : * Check if the start segment number is zero; this indicates a request to read
36 : : * any WAL file.
37 : : */
38 : : #define READ_ANY_WAL(privateInfo) ((privateInfo)->start_segno == 0)
39 : :
40 : : /*
41 : : * Hash entry representing a WAL segment retrieved from the archive.
42 : : *
43 : : * While WAL segments are typically read sequentially, individual entries
44 : : * maintain their own buffers for the following reasons:
45 : : *
46 : : * 1. Boundary Handling: The archive streamer provides a continuous byte
47 : : * stream. A single streaming chunk may contain the end of one WAL segment
48 : : * and the start of the next. Separate buffers allow us to easily
49 : : * partition and track these bytes by their respective segments.
50 : : *
51 : : * 2. Out-of-Order Support: Dedicated buffers simplify logic when segments
52 : : * are archived or retrieved out of sequence.
53 : : *
54 : : * To minimize the memory footprint, entries and their associated buffers are
55 : : * freed once consumed. Since pg_waldump does not request the same bytes
56 : : * twice (after it's located the point at which it should start decoding),
57 : : * a segment can be discarded as soon as pg_waldump moves past it. Moreover,
58 : : * if we read a segment that won't be needed till later, we spill its data to
59 : : * a temporary file instead of retaining it in memory. This ensures that
60 : : * pg_waldump can process even very large tar archives without needing more
61 : : * than a few WAL segments' worth of memory space.
62 : : */
63 : : typedef struct ArchivedWALFile
64 : : {
65 : : uint32 status; /* hash status */
66 : : const char *fname; /* hash key: WAL segment name */
67 : :
68 : : StringInfo buf; /* holds WAL bytes read from archive */
69 : : bool spilled; /* true if the WAL data was spilled to a
70 : : * temporary file */
71 : :
72 : : int read_len; /* total bytes received from archive for this
73 : : * segment (same as buf->len, unless we have
74 : : * spilled the data to a temp file) */
75 : : } ArchivedWALFile;
76 : :
77 : : static uint32 hash_string_pointer(const char *s);
78 : : #define SH_PREFIX ArchivedWAL
79 : : #define SH_ELEMENT_TYPE ArchivedWALFile
80 : : #define SH_KEY_TYPE const char *
81 : : #define SH_KEY fname
82 : : #define SH_HASH_KEY(tb, key) hash_string_pointer(key)
83 : : #define SH_EQUAL(tb, a, b) (strcmp(a, b) == 0)
84 : : #define SH_SCOPE static inline
85 : : #define SH_RAW_ALLOCATOR pg_malloc0
86 : : #define SH_DECLARE
87 : : #define SH_DEFINE
88 : : #include "lib/simplehash.h"
89 : :
90 : : typedef struct astreamer_waldump
91 : : {
92 : : astreamer base;
93 : : XLogDumpPrivate *privateInfo;
94 : : } astreamer_waldump;
95 : :
96 : : static ArchivedWALFile *get_archive_wal_entry(const char *fname,
97 : : XLogDumpPrivate *privateInfo);
98 : : static bool read_archive_file(XLogDumpPrivate *privateInfo);
99 : : static void setup_tmpwal_dir(const char *waldir);
100 : :
101 : : static FILE *prepare_tmp_write(const char *fname, XLogDumpPrivate *privateInfo);
102 : : static void perform_tmp_write(const char *fname, StringInfo buf, FILE *file);
103 : :
104 : : static astreamer *astreamer_waldump_new(XLogDumpPrivate *privateInfo);
105 : : static void astreamer_waldump_content(astreamer *streamer,
106 : : astreamer_member *member,
107 : : const char *data, int len,
108 : : astreamer_archive_context context);
109 : : static void astreamer_waldump_finalize(astreamer *streamer);
110 : : static void astreamer_waldump_free(astreamer *streamer);
111 : :
112 : : static bool member_is_wal_file(astreamer_waldump *mystreamer,
113 : : astreamer_member *member,
114 : : char **fname);
115 : :
116 : : static const astreamer_ops astreamer_waldump_ops = {
117 : : .content = astreamer_waldump_content,
118 : : .finalize = astreamer_waldump_finalize,
119 : : .free = astreamer_waldump_free
120 : : };
121 : :
122 : : /*
123 : : * Initializes the tar archive reader: opens the archive, builds a hash table
124 : : * for WAL entries, reads ahead until a full WAL page header is available to
125 : : * determine the WAL segment size, and computes start/end segment numbers for
126 : : * filtering.
127 : : */
128 : : void
129 : 52 : init_archive_reader(XLogDumpPrivate *privateInfo,
130 : : pg_compress_algorithm compression)
131 : : {
132 : : int fd;
133 : : astreamer *streamer;
134 : 52 : ArchivedWALFile *entry = NULL;
135 : : XLogLongPageHeader longhdr;
136 : : ArchivedWAL_iterator iter;
137 : :
138 : : /* Open tar archive and store its file descriptor */
139 : 52 : fd = open_file_in_directory(privateInfo->archive_dir,
140 : 52 : privateInfo->archive_name);
141 : :
142 [ - + ]: 52 : if (fd < 0)
143 : 0 : pg_fatal("could not open file \"%s\"", privateInfo->archive_name);
144 : :
145 : 52 : privateInfo->archive_fd = fd;
146 : 52 : privateInfo->archive_fd_eof = false;
147 : :
148 : 52 : streamer = astreamer_waldump_new(privateInfo);
149 : :
150 : : /* We must first parse the tar archive. */
151 : 52 : streamer = astreamer_tar_parser_new(streamer);
152 : :
153 : : /* If the archive is compressed, decompress before parsing. */
154 [ + + ]: 52 : if (compression == PG_COMPRESSION_GZIP)
155 : 17 : streamer = astreamer_gzip_decompressor_new(streamer);
156 [ + + ]: 35 : else if (compression == PG_COMPRESSION_LZ4)
157 : 4 : streamer = astreamer_lz4_decompressor_new(streamer);
158 [ - + ]: 31 : else if (compression == PG_COMPRESSION_ZSTD)
159 : 0 : streamer = astreamer_zstd_decompressor_new(streamer);
160 : :
161 : 52 : privateInfo->archive_streamer = streamer;
162 : :
163 : : /*
164 : : * Allocate a buffer for reading the archive file to begin content
165 : : * decoding.
166 : : */
167 : 52 : privateInfo->archive_read_buf = pg_malloc(READ_CHUNK_SIZE);
168 : 52 : privateInfo->archive_read_buf_size = READ_CHUNK_SIZE;
169 : :
170 : : /*
171 : : * Hash table storing WAL entries read from the archive with an arbitrary
172 : : * initial size.
173 : : */
174 : 52 : privateInfo->archive_wal_htab = ArchivedWAL_create(8, NULL);
175 : :
176 : : /*
177 : : * Read until we have at least one WAL segment with enough data to extract
178 : : * the WAL segment size from the long page header.
179 : : *
180 : : * We must not rely on cur_file here, because it can become NULL if a
181 : : * member trailer is processed during a read_archive_file() call. Instead,
182 : : * scan the hash table after each read to find any entry with sufficient
183 : : * data.
184 : : */
185 [ + + ]: 694 : while (entry == NULL)
186 : : {
187 [ - + ]: 642 : if (!read_archive_file(privateInfo))
188 : 0 : pg_fatal("could not find WAL in archive \"%s\"",
189 : : privateInfo->archive_name);
190 : :
191 : 642 : ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
192 : 642 : while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
193 [ + + ]: 642 : &iter)) != NULL)
194 : : {
195 [ + - ]: 52 : if (entry->read_len >= sizeof(XLogLongPageHeaderData))
196 : 52 : break;
197 : : }
198 : : }
199 : :
200 : : /* Extract the WAL segment size from the long page header */
201 : 52 : longhdr = (XLogLongPageHeader) entry->buf->data;
202 : :
203 [ + - + - : 52 : if (!IsValidWalSegSize(longhdr->xlp_seg_size))
+ - - + ]
204 : : {
205 : 0 : pg_log_error(ngettext("invalid WAL segment size in WAL file from archive \"%s\" (%u byte)",
206 : : "invalid WAL segment size in WAL file from archive \"%s\" (%u bytes)",
207 : : longhdr->xlp_seg_size),
208 : : privateInfo->archive_name, longhdr->xlp_seg_size);
209 : 0 : pg_log_error_detail("The WAL segment size must be a power of two between 1 MB and 1 GB.");
210 : 0 : exit(1);
211 : : }
212 : :
213 : 52 : privateInfo->segsize = longhdr->xlp_seg_size;
214 : :
215 : : /*
216 : : * With the WAL segment size available, we can now initialize the
217 : : * dependent start and end segment numbers.
218 : : */
219 : : Assert(XLogRecPtrIsValid(privateInfo->startptr));
220 : 52 : XLByteToSeg(privateInfo->startptr, privateInfo->start_segno,
221 : : privateInfo->segsize);
222 : :
223 [ + + ]: 52 : if (XLogRecPtrIsValid(privateInfo->endptr))
224 : 48 : XLByteToSeg(privateInfo->endptr, privateInfo->end_segno,
225 : : privateInfo->segsize);
226 : :
227 : : /*
228 : : * Now that we have initialized the filtering parameters (start_segno and
229 : : * end_segno), we can discard any already-loaded WAL hash table entries
230 : : * for segments we don't actually need. Subsequent WAL will be filtered
231 : : * automatically by the archive streamer using the updated start_segno and
232 : : * end_segno values.
233 : : */
234 : 52 : ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
235 : 119 : while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
236 [ + + ]: 119 : &iter)) != NULL)
237 : : {
238 : : XLogSegNo segno;
239 : : TimeLineID timeline;
240 : :
241 : 67 : XLogFromFileName(entry->fname, &timeline, &segno, privateInfo->segsize);
242 [ + - ]: 67 : if (privateInfo->timeline != timeline ||
243 [ + - ]: 67 : privateInfo->start_segno > segno ||
244 [ + + ]: 67 : privateInfo->end_segno < segno)
245 : 13 : free_archive_wal_entry(entry->fname, privateInfo);
246 : : }
247 : 52 : }
248 : :
249 : : /*
250 : : * Release the archive streamer chain and close the archive file.
251 : : */
252 : : void
253 : 48 : free_archive_reader(XLogDumpPrivate *privateInfo)
254 : : {
255 : : /*
256 : : * NB: Normally, astreamer_finalize() is called before astreamer_free() to
257 : : * flush any remaining buffered data or to ensure the end of the tar
258 : : * archive is reached. read_archive_file() may have done so. However,
259 : : * when decoding WAL we can stop once we hit the end LSN, so we may never
260 : : * have read all of the input file. In that case any remaining buffered
261 : : * data or unread portion of the archive can be safely ignored.
262 : : */
263 : 48 : astreamer_free(privateInfo->archive_streamer);
264 : :
265 : : /* Free any remaining hash table entries and their buffers. */
266 [ + - ]: 48 : if (privateInfo->archive_wal_htab != NULL)
267 : : {
268 : : ArchivedWAL_iterator iter;
269 : : ArchivedWALFile *entry;
270 : :
271 : 48 : ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
272 : 146 : while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
273 [ + + ]: 98 : &iter)) != NULL)
274 : : {
275 [ - + ]: 50 : if (entry->buf != NULL)
276 : 50 : destroyStringInfo(entry->buf);
277 : : }
278 : 48 : ArchivedWAL_destroy(privateInfo->archive_wal_htab);
279 : 48 : privateInfo->archive_wal_htab = NULL;
280 : : }
281 : :
282 : : /* Free the reusable read buffer. */
283 [ + - ]: 48 : if (privateInfo->archive_read_buf != NULL)
284 : : {
285 : 48 : pg_free(privateInfo->archive_read_buf);
286 : 48 : privateInfo->archive_read_buf = NULL;
287 : : }
288 : :
289 : : /* Close the file. */
290 [ - + ]: 48 : if (close(privateInfo->archive_fd) != 0)
291 : 0 : pg_log_error("could not close file \"%s\": %m",
292 : : privateInfo->archive_name);
293 : 48 : }
294 : :
295 : : /*
296 : : * Copies the requested WAL data from the hash entry's buffer into readBuff.
297 : : * If the buffer does not yet contain the needed bytes, fetches more data from
298 : : * the tar archive via the archive streamer.
299 : : */
300 : : int
301 : 28427 : read_archive_wal_page(XLogDumpPrivate *privateInfo, XLogRecPtr targetPagePtr,
302 : : size_t count, char *readBuff)
303 : : {
304 : 28427 : char *p = readBuff;
305 : 28427 : size_t nbytes = count;
306 : 28427 : XLogRecPtr recptr = targetPagePtr;
307 : 28427 : int segsize = privateInfo->segsize;
308 : : XLogSegNo segno;
309 : : char fname[MAXFNAMELEN];
310 : : ArchivedWALFile *entry;
311 : :
312 : : /* Identify the segment and locate its entry in the archive hash */
313 : 28427 : XLByteToSeg(targetPagePtr, segno, segsize);
314 : 28427 : XLogFileName(fname, privateInfo->timeline, segno, segsize);
315 : 28427 : entry = get_archive_wal_entry(fname, privateInfo);
316 : : Assert(!entry->spilled);
317 : :
318 [ + + ]: 60167 : while (nbytes > 0)
319 : : {
320 : 31740 : char *buf = entry->buf->data;
321 : 31740 : int bufLen = entry->buf->len;
322 : : XLogRecPtr endPtr;
323 : : XLogRecPtr startPtr;
324 : :
325 : : /*
326 : : * Calculate the LSN range currently residing in the buffer.
327 : : *
328 : : * read_len tracks total bytes received for this segment, so endPtr is
329 : : * the LSN just past the last buffered byte, and startPtr is the LSN
330 : : * of the first buffered byte.
331 : : */
332 : 31740 : XLogSegNoOffsetToRecPtr(segno, entry->read_len, segsize, endPtr);
333 : 31740 : startPtr = endPtr - bufLen;
334 : :
335 : : /*
336 : : * Copy the requested WAL record if it exists in the buffer.
337 : : */
338 [ + - + - : 31740 : if (bufLen > 0 && startPtr <= recptr && recptr < endPtr)
+ + ]
339 : 29470 : {
340 : : int copyBytes;
341 : 29470 : int offset = recptr - startPtr;
342 : :
343 : : /*
344 : : * Given startPtr <= recptr < endPtr and a total buffer size
345 : : * 'bufLen', the offset (recptr - startPtr) will always be less
346 : : * than 'bufLen'.
347 : : */
348 : : Assert(offset < bufLen);
349 : :
350 : 29470 : copyBytes = Min(nbytes, bufLen - offset);
351 : 29470 : memcpy(p, buf + offset, copyBytes);
352 : :
353 : : /* Update state for read */
354 : 29470 : recptr += copyBytes;
355 : 29470 : nbytes -= copyBytes;
356 : 29470 : p += copyBytes;
357 : : }
358 : : else
359 : : {
360 : : /*
361 : : * We evidently need to fetch more data. Raise an error if the
362 : : * archive streamer has moved past our segment (meaning the WAL
363 : : * file in the archive is shorter than expected) or if reading the
364 : : * archive reached EOF.
365 : : */
366 [ - + ]: 2270 : if (privateInfo->cur_file != entry)
367 : 0 : pg_fatal("WAL segment \"%s\" in archive \"%s\" is too short: read %zu of %zu bytes",
368 : : fname, privateInfo->archive_name,
369 : : (count - nbytes), count);
370 [ - + ]: 2270 : if (!read_archive_file(privateInfo))
371 : 0 : pg_fatal("unexpected end of archive \"%s\" while reading \"%s\": read %zu of %zu bytes",
372 : : privateInfo->archive_name, fname,
373 : : (count - nbytes), count);
374 : :
375 : : /*
376 : : * Loading more data may have moved hash table entries, so we must
377 : : * re-look-up the one we are reading from.
378 : : */
379 : 2270 : entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
380 : : /* ... it had better still be there */
381 : : Assert(entry != NULL);
382 : : }
383 : : }
384 : :
385 : : /*
386 : : * Should have successfully read all the requested bytes or reported a
387 : : * failure before this point.
388 : : */
389 : : Assert(nbytes == 0);
390 : :
391 : : /*
392 : : * Return count unchanged; the caller expects this convention, matching
393 : : * the routine that reads WAL pages from physical files.
394 : : */
395 : 28427 : return count;
396 : : }
397 : :
398 : : /*
399 : : * Releases the buffer of a WAL entry that is no longer needed, preventing the
400 : : * accumulation of irrelevant WAL data. Also removes any associated temporary
401 : : * file and clears privateInfo->cur_file if it points to this entry, so the
402 : : * archive streamer skips subsequent data for it.
403 : : */
404 : : void
405 : 43 : free_archive_wal_entry(const char *fname, XLogDumpPrivate *privateInfo)
406 : : {
407 : : ArchivedWALFile *entry;
408 : : const char *oldfname;
409 : :
410 : 43 : entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
411 : :
412 [ - + ]: 43 : if (entry == NULL)
413 : 0 : return;
414 : :
415 : : /* Destroy the buffer */
416 : 43 : destroyStringInfo(entry->buf);
417 : 43 : entry->buf = NULL;
418 : :
419 : : /* Remove temporary file if any */
420 [ + + ]: 43 : if (entry->spilled)
421 : : {
422 : : char fpath[MAXPGPATH];
423 : :
424 : 1 : snprintf(fpath, MAXPGPATH, "%s/%s", TmpWalSegDir, fname);
425 : :
426 [ + - ]: 1 : if (unlink(fpath) == 0)
427 [ - + ]: 1 : pg_log_debug("removed file \"%s\"", fpath);
428 : : }
429 : :
430 : : /* Clear cur_file if it points to the entry being freed */
431 [ + + ]: 43 : if (privateInfo->cur_file == entry)
432 : 4 : privateInfo->cur_file = NULL;
433 : :
434 : : /*
435 : : * ArchivedWAL_delete_item may cause other hash table entries to move.
436 : : * Therefore, if cur_file isn't NULL now, we have to be prepared to look
437 : : * that entry up again after the deletion. Fortunately, the entry's fname
438 : : * string won't move.
439 : : */
440 [ + + ]: 43 : oldfname = privateInfo->cur_file ? privateInfo->cur_file->fname : NULL;
441 : :
442 : 43 : ArchivedWAL_delete_item(privateInfo->archive_wal_htab, entry);
443 : :
444 [ + + ]: 43 : if (oldfname)
445 : : {
446 : 29 : privateInfo->cur_file = ArchivedWAL_lookup(privateInfo->archive_wal_htab,
447 : : oldfname);
448 : : /* ... it had better still be there */
449 : : Assert(privateInfo->cur_file != NULL);
450 : : }
451 : : }
452 : :
453 : : /*
454 : : * Returns the archived WAL entry from the hash table if it already exists.
455 : : * Otherwise, reads more data from the archive until the requested entry is
456 : : * found. If the archive streamer reads a WAL file from the archive that
457 : : * is not currently needed, that data is spilled to a temporary file for later
458 : : * retrieval.
459 : : *
460 : : * Note that the returned entry might not have been completely read from
461 : : * the archive yet.
462 : : */
463 : : static ArchivedWALFile *
464 : 28427 : get_archive_wal_entry(const char *fname, XLogDumpPrivate *privateInfo)
465 : : {
466 : : while (1)
467 : 1792 : {
468 : : ArchivedWALFile *entry;
469 : : ArchivedWAL_iterator iter;
470 : :
471 : : /*
472 : : * Search the hash table first. If the entry is found, return it.
473 : : * Otherwise, the requested WAL entry hasn't been read from the
474 : : * archive yet; we must invoke the archive streamer to fetch it.
475 : : */
476 : 30219 : entry = ArchivedWAL_lookup(privateInfo->archive_wal_htab, fname);
477 : :
478 [ + + ]: 30219 : if (entry != NULL)
479 : 28427 : return entry;
480 : :
481 : : /*
482 : : * Before loading more data, scan the hash table to see if we have
483 : : * loaded any files we don't need yet. If so, spill their data to
484 : : * disk to conserve memory space. But don't try to spill a
485 : : * partially-read file; it's not worth the complication.
486 : : */
487 : 1792 : ArchivedWAL_start_iterate(privateInfo->archive_wal_htab, &iter);
488 : 2304 : while ((entry = ArchivedWAL_iterate(privateInfo->archive_wal_htab,
489 [ + + ]: 2304 : &iter)) != NULL)
490 : : {
491 : : FILE *write_fp;
492 : :
493 : : /* OK to spill? */
494 [ + + ]: 512 : if (entry->spilled)
495 : 254 : continue; /* already spilled */
496 [ + + ]: 258 : if (entry == privateInfo->cur_file)
497 : 256 : continue; /* still being read */
498 : :
499 : : /* Write out the completed WAL file contents to a temp file. */
500 : 2 : write_fp = prepare_tmp_write(entry->fname, privateInfo);
501 : 2 : perform_tmp_write(entry->fname, entry->buf, write_fp);
502 [ - + ]: 2 : if (fclose(write_fp) != 0)
503 : 0 : pg_fatal("could not close file \"%s/%s\": %m",
504 : : TmpWalSegDir, entry->fname);
505 : :
506 : : /* resetStringInfo won't release storage, so delete/recreate. */
507 : 2 : destroyStringInfo(entry->buf);
508 : 2 : entry->buf = makeStringInfo();
509 : 2 : entry->spilled = true;
510 : : }
511 : :
512 : : /*
513 : : * Read more data. If we reach EOF, the desired file is not present.
514 : : */
515 [ - + ]: 1792 : if (!read_archive_file(privateInfo))
516 : 0 : pg_fatal("could not find WAL \"%s\" in archive \"%s\"",
517 : : fname, privateInfo->archive_name);
518 : : }
519 : : }
520 : :
521 : : /*
522 : : * Reads a chunk from the archive file and passes it through the streamer
523 : : * pipeline for decompression (if needed) and tar member extraction.
524 : : *
525 : : * Returns true if successful, false if there is no more data.
526 : : *
527 : : * Callers must be aware that a single call may trigger multiple callbacks
528 : : * in astreamer_waldump_content, so privateInfo->cur_file can change value
529 : : * (or become NULL) during a call. In particular, cur_file is set to NULL
530 : : * when the ASTREAMER_MEMBER_TRAILER callback fires at the end of a tar
531 : : * member; it is then set to a new entry when the next WAL member's
532 : : * ASTREAMER_MEMBER_HEADER callback fires, which may or may not happen
533 : : * within the same call.
534 : : */
535 : : static bool
536 : 4704 : read_archive_file(XLogDumpPrivate *privateInfo)
537 : : {
538 : : ssize_t rc;
539 : :
540 : : /* Fail if we already reached EOF in a prior call. */
541 [ - + ]: 4704 : if (privateInfo->archive_fd_eof)
542 : 0 : return false;
543 : :
544 : : /* Try to read some more data. */
545 : 4704 : rc = read(privateInfo->archive_fd, privateInfo->archive_read_buf,
546 : : privateInfo->archive_read_buf_size);
547 [ - + ]: 4704 : if (rc < 0)
548 : 0 : pg_fatal("could not read file \"%s\": %m",
549 : : privateInfo->archive_name);
550 : :
551 : : /*
552 : : * Decompress (if required), and then parse the previously read contents
553 : : * of the tar file.
554 : : */
555 [ + - ]: 4704 : if (rc > 0)
556 : 4704 : astreamer_content(privateInfo->archive_streamer, NULL,
557 : 4704 : privateInfo->archive_read_buf, rc,
558 : : ASTREAMER_UNKNOWN);
559 : : else
560 : : {
561 : : /*
562 : : * We reached EOF, but there is probably still data queued in the
563 : : * astreamer pipeline's buffers. Flush it out to ensure that we
564 : : * process everything.
565 : : */
566 : 0 : astreamer_finalize(privateInfo->archive_streamer);
567 : : /* Set flag to ensure we don't finalize more than once. */
568 : 0 : privateInfo->archive_fd_eof = true;
569 : : }
570 : :
571 : 4704 : return true;
572 : : }
573 : :
574 : : /*
575 : : * Set up a temporary directory to temporarily store WAL segments.
576 : : */
577 : : static void
578 : 2 : setup_tmpwal_dir(const char *waldir)
579 : : {
580 : 2 : const char *tmpdir = getenv("TMPDIR");
581 : : char *template;
582 : :
583 : : Assert(TmpWalSegDir == NULL);
584 : :
585 : : /*
586 : : * Use the directory specified by the TMPDIR environment variable. If it's
587 : : * not set, fall back to the provided WAL directory to store WAL files
588 : : * temporarily.
589 : : */
590 [ - + ]: 2 : template = psprintf("%s/waldump_tmp-XXXXXX",
591 : : tmpdir ? tmpdir : waldir);
592 : 2 : TmpWalSegDir = mkdtemp(template);
593 : :
594 [ - + ]: 2 : if (TmpWalSegDir == NULL)
595 : 0 : pg_fatal("could not create directory \"%s\": %m", template);
596 : :
597 : 2 : canonicalize_path(TmpWalSegDir);
598 : :
599 [ - + ]: 2 : pg_log_debug("created directory \"%s\"", TmpWalSegDir);
600 : 2 : }
601 : :
602 : : /*
603 : : * Open a file in the temporary spill directory for writing an out-of-order
604 : : * WAL segment, creating the directory if not already done.
605 : : * Returns the open file handle.
606 : : */
607 : : static FILE *
608 : 2 : prepare_tmp_write(const char *fname, XLogDumpPrivate *privateInfo)
609 : : {
610 : : char fpath[MAXPGPATH];
611 : : FILE *file;
612 : :
613 : : /* Setup temporary directory to store WAL segments, if we didn't already */
614 [ + - ]: 2 : if (unlikely(TmpWalSegDir == NULL))
615 : 2 : setup_tmpwal_dir(privateInfo->archive_dir);
616 : :
617 : 2 : snprintf(fpath, MAXPGPATH, "%s/%s", TmpWalSegDir, fname);
618 : :
619 : : /* Open the spill file for writing */
620 : 2 : file = fopen(fpath, PG_BINARY_W);
621 [ - + ]: 2 : if (file == NULL)
622 : 0 : pg_fatal("could not create file \"%s\": %m", fpath);
623 : :
624 : : #ifndef WIN32
625 [ - + ]: 2 : if (chmod(fpath, pg_file_create_mode))
626 : 0 : pg_fatal("could not set permissions on file \"%s\": %m",
627 : : fpath);
628 : : #endif
629 : :
630 [ - + ]: 2 : pg_log_debug("spilling to temporary file \"%s\"", fpath);
631 : :
632 : 2 : return file;
633 : : }
634 : :
635 : : /*
636 : : * Write buffer data to the given file handle.
637 : : */
638 : : static void
639 : 2 : perform_tmp_write(const char *fname, StringInfo buf, FILE *file)
640 : : {
641 : : Assert(file);
642 : :
643 : 2 : errno = 0;
644 [ + - - + ]: 2 : if (buf->len > 0 && fwrite(buf->data, buf->len, 1, file) != 1)
645 : : {
646 : : /*
647 : : * If write didn't set errno, assume problem is no disk space
648 : : */
649 [ # # ]: 0 : if (errno == 0)
650 : 0 : errno = ENOSPC;
651 : 0 : pg_fatal("could not write to file \"%s/%s\": %m", TmpWalSegDir, fname);
652 : : }
653 : 2 : }
654 : :
655 : : /*
656 : : * Create an astreamer that can read WAL from tar file.
657 : : */
658 : : static astreamer *
659 : 52 : astreamer_waldump_new(XLogDumpPrivate *privateInfo)
660 : : {
661 : : astreamer_waldump *streamer;
662 : :
663 : 52 : streamer = palloc0_object(astreamer_waldump);
664 : 52 : *((const astreamer_ops **) &streamer->base.bbs_ops) =
665 : : &astreamer_waldump_ops;
666 : :
667 : 52 : streamer->privateInfo = privateInfo;
668 : :
669 : 52 : return &streamer->base;
670 : : }
671 : :
672 : : /*
673 : : * Main entry point of the archive streamer for reading WAL data from a tar
674 : : * file. If a member is identified as a valid WAL file, a hash entry is created
675 : : * for it, and its contents are copied into that entry's buffer, making them
676 : : * accessible to the decoding routine.
677 : : */
678 : : static void
679 : 31942 : astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
680 : : const char *data, int len,
681 : : astreamer_archive_context context)
682 : : {
683 : 31942 : astreamer_waldump *mystreamer = (astreamer_waldump *) streamer;
684 : 31942 : XLogDumpPrivate *privateInfo = mystreamer->privateInfo;
685 : :
686 : : Assert(context != ASTREAMER_UNKNOWN);
687 : :
688 [ + + + - : 31942 : switch (context)
- ]
689 : : {
690 : 8699 : case ASTREAMER_MEMBER_HEADER:
691 : : {
692 : 8699 : char *fname = NULL;
693 : : ArchivedWALFile *entry;
694 : : bool found;
695 : :
696 : : /* Shouldn't see MEMBER_HEADER in the middle of a file */
697 : : Assert(privateInfo->cur_file == NULL);
698 : :
699 [ - + ]: 8699 : pg_log_debug("reading \"%s\"", member->pathname);
700 : :
701 [ + + ]: 8699 : if (!member_is_wal_file(mystreamer, member, &fname))
702 : 8602 : break;
703 : :
704 : : /*
705 : : * Skip range filtering during initial startup, before the WAL
706 : : * segment size and segment number bounds are known.
707 : : */
708 [ + + ]: 109 : if (!READ_ANY_WAL(privateInfo))
709 : : {
710 : : XLogSegNo segno;
711 : : TimeLineID timeline;
712 : :
713 : : /*
714 : : * Skip the segment if the timeline does not match, if it
715 : : * falls outside the caller-specified range.
716 : : */
717 : 42 : XLogFromFileName(fname, &timeline, &segno, privateInfo->segsize);
718 [ + - ]: 42 : if (privateInfo->timeline != timeline ||
719 [ + - ]: 42 : privateInfo->start_segno > segno ||
720 [ + + ]: 42 : privateInfo->end_segno < segno)
721 : : {
722 : 12 : pfree(fname);
723 : 12 : break;
724 : : }
725 : : }
726 : :
727 : : /*
728 : : * Note: ArchivedWAL_insert may cause existing hash table
729 : : * entries to move. While cur_file is known to be NULL right
730 : : * now, read_archive_wal_page may have a live hash entry
731 : : * pointer, which it needs to take care to update after
732 : : * read_archive_file completes.
733 : : */
734 : 97 : entry = ArchivedWAL_insert(privateInfo->archive_wal_htab,
735 : : fname, &found);
736 : :
737 : : /*
738 : : * Shouldn't happen, but if it does, simply ignore the
739 : : * duplicate WAL file.
740 : : */
741 [ - + ]: 97 : if (found)
742 : : {
743 : 0 : pg_log_warning("ignoring duplicate WAL \"%s\" found in archive \"%s\"",
744 : : member->pathname, privateInfo->archive_name);
745 : 0 : pfree(fname);
746 : 0 : break;
747 : : }
748 : :
749 : 97 : entry->buf = makeStringInfo();
750 : 97 : entry->spilled = false;
751 : 97 : entry->read_len = 0;
752 : 97 : privateInfo->cur_file = entry;
753 : : }
754 : 97 : break;
755 : :
756 : 14596 : case ASTREAMER_MEMBER_CONTENTS:
757 [ + + ]: 14596 : if (privateInfo->cur_file)
758 : : {
759 : 5388 : appendBinaryStringInfo(privateInfo->cur_file->buf, data, len);
760 : 5388 : privateInfo->cur_file->read_len += len;
761 : : }
762 : 14596 : break;
763 : :
764 : 8647 : case ASTREAMER_MEMBER_TRAILER:
765 : :
766 : : /*
767 : : * End of this tar member; mark cur_file NULL so subsequent
768 : : * content callbacks (if any) know no WAL file is currently
769 : : * active.
770 : : */
771 : 8647 : privateInfo->cur_file = NULL;
772 : 8647 : break;
773 : :
774 : 0 : case ASTREAMER_ARCHIVE_TRAILER:
775 : 0 : break;
776 : :
777 : 0 : default:
778 : : /* Shouldn't happen. */
779 : 0 : pg_fatal("unexpected state while parsing tar file");
780 : : }
781 : 31942 : }
782 : :
783 : : /*
784 : : * End-of-stream processing for an astreamer_waldump stream. This is a
785 : : * terminal streamer so it must have no successor.
786 : : */
787 : : static void
788 : 0 : astreamer_waldump_finalize(astreamer *streamer)
789 : : {
790 : : Assert(streamer->bbs_next == NULL);
791 : 0 : }
792 : :
793 : : /*
794 : : * Free memory associated with an astreamer_waldump stream.
795 : : */
796 : : static void
797 : 48 : astreamer_waldump_free(astreamer *streamer)
798 : : {
799 : : Assert(streamer->bbs_next == NULL);
800 : 48 : pfree(streamer);
801 : 48 : }
802 : :
803 : : /*
804 : : * Returns true if the archive member name matches the WAL naming format. If
805 : : * successful, it also outputs the WAL segment name.
806 : : */
807 : : static bool
808 : 8699 : member_is_wal_file(astreamer_waldump *mystreamer, astreamer_member *member,
809 : : char **fname)
810 : : {
811 : : int pathlen;
812 : : char pathname[MAXPGPATH];
813 : : char *filename;
814 : :
815 : : /* We are only interested in normal files */
816 [ + + ]: 8699 : if (!member->is_regular)
817 : 282 : return false;
818 : :
819 [ + + ]: 8417 : if (strlen(member->pathname) < XLOG_FNAME_LEN)
820 : 8225 : return false;
821 : :
822 : : /*
823 : : * For a correct comparison, we must remove any '.' or '..' components
824 : : * from the member pathname. Similar to member_verify_header(), we prepend
825 : : * './' to the path so that canonicalize_path() can properly resolve and
826 : : * strip these references from the tar member name.
827 : : */
828 : 192 : snprintf(pathname, MAXPGPATH, "./%s", member->pathname);
829 : 192 : canonicalize_path(pathname);
830 : 192 : pathlen = strlen(pathname);
831 : :
832 : : /* Skip files in subdirectories other than pg_wal/ */
833 [ + + ]: 192 : if (pathlen > XLOG_FNAME_LEN &&
834 [ + + ]: 91 : strncmp(pathname, XLOGDIR, strlen(XLOGDIR)) != 0)
835 : 83 : return false;
836 : :
837 : : /* WAL file may appear with a full path (e.g., pg_wal/<name>) */
838 : 109 : filename = pathname + (pathlen - XLOG_FNAME_LEN);
839 [ - + ]: 109 : if (!IsXLogFileName(filename))
840 : 0 : return false;
841 : :
842 : 109 : *fname = pnstrdup(filename, XLOG_FNAME_LEN);
843 : :
844 : 109 : return true;
845 : : }
846 : :
847 : : /*
848 : : * Helper function for WAL file hash table.
849 : : */
850 : : static uint32
851 : 32701 : hash_string_pointer(const char *s)
852 : : {
853 : 32701 : const unsigned char *ss = (const unsigned char *) s;
854 : :
855 : 32701 : return hash_bytes(ss, strlen(s));
856 : : }
|
