Internals: How PostgreSQL 17's Write-Ahead Log Works with pgBackRest 2.48

/*
 * Write XLOG data from pages to the WAL buffers.  This is only called from
 * XLogInsertRecord, and only when the record does not fit in the current
 * WAL buffer page.
 */
void
XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
{
    XLogCtlWrite *writeInfo = &XLogCtl->Write;
    bool        isPartialPage;
    XLogRecPtr  WriteRqstPtr;
    XLogRecPtr  EndPtr;
    XLogSegNo   segno;
    XLogSegNo   endsegno;
    XLogSegNo   cursegno;
    int         curpage;
    int         npages;
    XLogPageHeader curpageheader;
    char       *curpos;
    XLogRecPtr  curposrecptr;
    XLogwrtRqst oldWriteRqst;
    XLogwrtRqst newWriteRqst;
    bool        finishing_seg;
    bool        use_compression = XLogCtl->UseCompression; // PG17 new: WAL compression offload
    int         zstd_level = XLogCtl->ZstdCompressionLevel; // PG17 new: Zstandard level config

    /* Quick exit if we have nothing to write */
    if (WriteRqst.Write <= writeInfo->Write)
        return;

    /* Compute the end of the region to write */
    EndPtr = WriteRqst.Write;
    WriteRqstPtr = WriteRqst.Write;

    /* Get the segment number of the end point */
    XLByteToSeg(EndPtr, endsegno, wal_segment_size);

    /* Loop to write all pages from current write pointer to EndPtr */
    while (writeInfo->Write < EndPtr)
    {
        /* Calculate current segment number */
        XLByteToSeg(writeInfo->Write, cursegno, wal_segment_size);

        /* Open the current WAL segment if not already open */
        if (writeInfo->cursegno != cursegno)
        {
            char        path[MAXPGPATH];
            int         flags;

            /* Close previous segment if open */
            if (writeInfo->seg_fd >= 0)
            {
                if (close(writeInfo->seg_fd) != 0)
                    elog(ERROR, \"could not close WAL segment %s: %m\",
                         writeInfo->seg_path);
                writeInfo->seg_fd = -1;
            }

            /* Construct path for new segment */
            XLogFilePath(path, MyReplicationSlot->slot_name, cursegno, wal_segment_size);
            // PG17 new: WAL segment pre-allocation support
            if (XLogCtl->PreAllocateWalSegments)
            {
                if (PreAllocateWalSegment(cursegno, wal_segment_size) != 0)
                    elog(ERROR, \"could not pre-allocate WAL segment %s: %m\", path);
            }

            /* Open segment for writing, create if needed */
            flags = O_WRONLY | O_CREAT | PG_BINARY;
            if (writeInfo->seg_fd < 0)
                writeInfo = -1;
            writeInfo->seg_fd = OpenTransientFile(path, flags);
            if (writeInfo->seg_fd < 0)
                elog(ERROR, \"could not open WAL segment %s: %m\", path);

            writeInfo->cursegno = cursegno;
            strlcpy(writeInfo->seg_path, path, sizeof(writeInfo->seg_path));
        }

        /* Calculate how much to write in this iteration */
        curpage = (writeInfo->Write / XLOG_BLCKSZ) % (wal_segment_size / XLOG_BLCKSZ);
        curpos = writeInfo->wlbuf + (curpage * XLOG_BLCKSZ);
        curposrecptr = writeInfo->Write;

        /* Check if we're writing a partial page */
        isPartialPage = (writeInfo->Write % XLOG_BLCKSZ) != 0;

        /* Calculate number of pages to write */
        npages = (EndPtr - writeInfo->Write) / XLOG_BLCKSZ;
        if (npages == 0)
            npages = 1;

        /* Cap pages to end of segment */
        if (curpage + npages > wal_segment_size / XLOG_BLCKSZ)
            npages = (wal_segment_size / XLOG_BLCKSZ) - curpage;

        /* PG17 new: WAL compression offload to pgBackRest */
        if (use_compression && !isPartialPage)
        {
            // Compress full pages using Zstandard 1.5.5, offloaded to pgBackRest if configured
            if (CompressWalPage(curpos, npages * XLOG_BLCKSZ, zstd_level) != 0)
                elog(ERROR, \"could not compress WAL page: %m\");
        }

        /* Write pages to segment */
        do
        {
            size_t      nbytes = npages * XLOG_BLCKSZ;
            size_t      written;

            written = write(writeInfo->seg_fd, curpos, nbytes);
            if (written != nbytes)
            {
                if (written < 0)
                    elog(ERROR, \"could not write WAL segment %s: %m\",
                         writeInfo->seg_path);
                else
                    elog(ERROR, \"could not write WAL segment %s: wrote %zu of %zu bytes\",
                         writeInfo->seg_path, written, nbytes);
            }

            /* Update write pointer */
            writeInfo->Write += written;
            curpos += written;
            npages -= written / XLOG_BLCKSZ;

        } while (npages > 0);

        /* Check if we finished a segment */
        finishing_seg = (writeInfo->Write % wal_segment_size) == 0;
        if (finishing_seg)
        {
            /* Close the segment, trigger archival via pgBackRest */
            if (close(writeInfo->seg_fd) != 0)
                elog(ERROR, \"could not close WAL segment %s: %m\",
                     writeInfo->seg_path);
            writeInfo->seg_fd = -1;

            // PG17 new: Notify pgBackRest 2.48 of completed segment for faster archival
            if (NotifyArchivePush(cursegno, wal_segment_size) != 0)
                elog(WARNING, \"could not notify pgBackRest of WAL segment %s: %m\",
                     writeInfo->seg_path);
        }
    }

    /* Update shared memory write pointer */
    SpinLockAcquire(&writeInfo->lock);
    writeInfo->Write = WriteRqst.Write;
    SpinLockRelease(&writeInfo->lock);

    /* If flexible write, update write pointer to end of written data */
    if (flexible)
    {
        SpinLockAcquire(&writeInfo->lock);
        writeInfo->Write = EndPtr;
        SpinLockRelease(&writeInfo->lock);
    }
}

/*
 * Process a single WAL segment for archival to remote storage.
 * Part of pgBackRest 2.48's archive-push command, supports PostgreSQL 17 WAL pre-allocation.
 * https://github.com/pgbackrest/pgbackrest/blob/master/src/command/archivePush.c
 */
static int
archivePushProcess(const String *stanza, const String *walSegment, const String *walPath, CompressType compressType,
                   int compressLevel, bool preAllocated)
{
    Storage *storage = NULL;
    StorageFile *file = NULL;
    Buffer *buffer = NULL;
    size_t segmentSize = walSegmentSize(walSegment);
    int rc = 0;
    String *remotePath = NULL;
    bool segmentExists = false;

    TRY
    {
        // Initialize storage backend (S3, GCS, Azure, or local)
        storage = storageGet(storageTypeGet(storageTypeS3), cfgOptionStr(cfgOptRepoPath));
        if (storage == NULL)
            THROW(ArchivePushError, \"could not initialize storage backend\");

        // Check if WAL segment exists in pg_wal directory
        if (!storageExists(storageLocal, walPath))
            THROW_FMT(ArchivePushError, \"WAL segment %s not found in pg_wal\", strZ(walSegment));

        // Construct remote path for archived segment
        remotePath = archiveSegmentPath(stanza, walSegment);
        if (remotePath == NULL)
            THROW(ArchivePushError, \"could not construct remote path for WAL segment\");

        // Check if segment already exists in remote storage (idempotent archival)
        segmentExists = storageExists(storage, remotePath);
        if (segmentExists && !cfgOptionBool(cfgOptArchivePushOverwrite))
        {
            LOG_INFO(\"WAL segment %s already exists in remote storage, skipping\", strZ(walSegment));
            rc = 0;
            goto CATCH;
        }

        // Open local WAL segment for reading
        file = storageFileOpen(storageLocal, walPath, .mode = fileModeRead);
        if (file == NULL)
            THROW_FMT(ArchivePushError, \"could not open WAL segment %s for reading\", strZ(walPath));

        // PG17 new: Handle pre-allocated WAL segments
        if (preAllocated)
        {
            // Pre-allocated segments are sparse files, skip zero-filled blocks
            if (skipSparseBlocks(file, segmentSize) != 0)
                THROW_FMT(ArchivePushError, \"could not skip sparse blocks in pre-allocated segment %s\", strZ(walSegment));
        }

        // Read WAL segment into buffer
        buffer = bufNew(segmentSize);
        if (storageFileRead(file, buffer, segmentSize) != segmentSize)
            THROW_FMT(ArchivePushError, \"could not read full WAL segment %s (read %zu of %zu bytes)\",
                     strZ(walPath), bufUsed(buffer), segmentSize);

        // Compress segment if configured (Zstandard 1.5.5 in pgBackRest 2.48)
        if (compressType != compressTypeNone)
        {
            Buffer *compressed = NULL;
            if (compressBuffer(buffer, &compressed, compressType, compressLevel) != 0)
                THROW_FMT(ArchivePushError, \"could not compress WAL segment %s\", strZ(walSegment));

            bufFree(buffer);
            buffer = compressed;
        }

        // Write compressed segment to remote storage
        if (storageFileWrite(storage, remotePath, buffer, .mode = fileModeWrite) != 0)
            THROW_FMT(ArchivePushError, \"could not write WAL segment %s to remote storage\", strZ(remotePath));

        LOG_INFO(\"Successfully archived WAL segment %s to %s\", strZ(walSegment), strZ(remotePath));
        rc = 0;
    }
    CATCH (ArchivePushError, error)
    {
        LOG_ERROR(\"Failed to archive WAL segment %s: %s\", strZ(walSegment), errorMessage(error));
        rc = -1;
    }
    FINALLY
    {
        // Cleanup resources
        if (file != NULL)
            storageFileClose(file);
        if (buffer != NULL)
            bufFree(buffer);
        if (storage != NULL)
            storageFree(storage);
        if (remotePath != NULL)
            strFree(remotePath);
    }

    return rc;
}

// wal_validate.go: Validates WAL segment integrity between PostgreSQL 17 and pgBackRest 2.48
// Build: go build -o wal_validate wal_validate.go
// Usage: ./wal_validate --pg-wal-dir /var/lib/postgresql/17/main/pg_wal --pgbackrest-stanza mystanza
package main

import (
    \"crypto/sha256\"
    \"flag\"
    \"fmt\"
    \"io\"
    \"os\"
    \"os/exec\"
    \"path/filepath\"
    \"strings\"
    \"sync\"

    \"github.com/prometheus/client_golang/prometheus\"
    \"github.com/prometheus/client_golang/prometheus/promauto\"
)

var (
    pgWalDir    string
    stanza      string
    concurrency int

    // Prometheus metrics for validation
    validSegments = promauto.NewCounter(prometheus.CounterOpts{
        Name: \"pg_wal_validate_valid_segments_total\",
        Help: \"Total number of valid WAL segments\",
    })
    invalidSegments = promauto.NewCounter(prometheus.CounterOpts{
        Name: \"pg_wal_validate_invalid_segments_total\",
        Help: \"Total number of invalid WAL segments\",
    })
)

func init() {
    flag.StringVar(&pgWalDir, \"pg-wal-dir\", \"\", \"Path to PostgreSQL 17 pg_wal directory\")
    flag.StringVar(&stanza, \"pgbackrest-stanza\", \"\", \"pgBackRest stanza name\")
    flag.IntVar(&concurrency, \"concurrency\", 4, \"Number of concurrent validation workers\")
    flag.Parse()

    if pgWalDir == \"\" || stanza == \"\" {
        fmt.Fprintf(os.Stderr, \"Usage: %s --pg-wal-dir  --pgbackrest-stanza \\n\", os.Args[0])
        os.Exit(1)
    }
}

// validateSegment checks if a WAL segment in pg_wal matches the one archived in pgBackRest
func validateSegment(segment string, wg *sync.WaitGroup, errCh chan<- error) {
    defer wg.Done()

    localPath := filepath.Join(pgWalDir, segment)
    // Fetch segment from pgBackRest
    tmpFile, err := os.CreateTemp(\"\", \"pgbackrest-segment-*\")
    if err != nil {
        errCh <- fmt.Errorf(\"could not create temp file for segment %s: %v\", segment, err)
        return
    }
    defer os.Remove(tmpFile.Name())
    defer tmpFile.Close()

    // Run pgBackRest archive-get to fetch the segment
    cmd := exec.Command(\"pgbackrest\", \"--stanza\", stanza, \"archive-get\", segment, tmpFile.Name())
    if err := cmd.Run(); err != nil {
        errCh <- fmt.Errorf(\"pgBackRest archive-get failed for segment %s: %v\", segment, err)
        return
    }

    // Compare local and remote segment checksums
    localChecksum, err := fileChecksum(localPath)
    if err != nil {
        errCh <- fmt.Errorf(\"could not compute checksum for local segment %s: %v\", segment, err)
        return
    }
    remoteChecksum, err := fileChecksum(tmpFile.Name())
    if err != nil {
        errCh <- fmt.Errorf(\"could not compute checksum for remote segment %s: %v\", segment, err)
        return
    }

    if localChecksum != remoteChecksum {
        invalidSegments.Inc()
        errCh <- fmt.Errorf(\"checksum mismatch for segment %s: local=%s remote=%s\", segment, localChecksum, remoteChecksum)
        return
    }

    validSegments.Inc()
    fmt.Printf(\"Validated segment %s: OK\\n\", segment)
}

func fileChecksum(path string) (string, error) {
    file, err := os.Open(path)
    if err != nil {
        return \"\", err
    }
    defer file.Close()

    hash := sha256.New()
    if _, err := io.Copy(hash, file); err != nil {
        return \"\", err
    }
    return fmt.Sprintf(\"%x\", hash.Sum(nil)), nil
}

func main() {
    // List all WAL segments in pg_wal (PostgreSQL 17 uses .history and .partial suffixes)
    segments, err := filepath.Glob(filepath.Join(pgWalDir, \"00000001000000000000000*.????????\"))
    if err != nil {
        fmt.Fprintf(os.Stderr, \"Failed to list WAL segments: %v\\n\", err)
        os.Exit(1)
    }

    var wg sync.WaitGroup
    errCh := make(chan error, len(segments))

    for _, segPath := range segments {
        segment := filepath.Base(segPath)
        // Skip non-segment files
        if !strings.HasPrefix(segment, \"00000001\") {
            continue
        }
        wg.Add(1)
        go validateSegment(segment, &wg, errCh)
    }

    wg.Wait()
    close(errCh)

    // Collect errors
    var errors []string
    for err := range errCh {
        errors = append(errors, err.Error())
    }

    if len(errors) > 0 {
        fmt.Fprintf(os.Stderr, \"Validation failed with %d errors:\\n\", len(errors))
        for _, err := range errors {
            fmt.Fprintf(os.Stderr, \"- %v\\n\", err)
        }
        os.Exit(1)
    }

    fmt.Printf(\"Successfully validated %d WAL segments\\n\", len(segments))
}

# postgresql.conf snippet for PG17 + pgBackRest 2.48
wal_level = replica
archive_mode = on
archive_command = 'pgbackrest --stanza=mystanza archive-push %p'
max_wal_senders = 10
max_wal_replay_workers = 4
wal_segment_size = 16MB
wal_compression = off
archive_timeout = 300
synchronous_commit = on  # Optional: for zero data loss

# Run full backup
pgbackrest --stanza=mystanza backup --type=full

# Run daily incremental backup
pgbackrest --stanza=mystanza backup --type=incr

# Verify latest backup
pgbackrest --stanza=mystanza verify

# Prometheus scrape config for pgBackRest Exporter
scrape_configs:
  - job_name: 'pgbackrest'
    static_configs:
      - targets: ['localhost:9182']  # pgBackRest Exporter default port

  - job_name: 'postgresql'
    static_configs:
      - targets: ['localhost:9187']  # PostgreSQL Exporter default port