Migrate page checksum filter to C.

This filter exactly mimics the behavior of the Perl filter so is a drop-in replacement.

The filter is not integrated yet since it requires the Perl-to-C storage layer interface coming in a future commit.

src/Makefile.in

@@ -214,6 +214,9 @@ command/archive/push/push.o: command/archive/push/push.c build.auto.h command/ar
 command/backup/common.o: command/backup/common.c build.auto.h command/backup/common.h common/assert.h common/debug.h common/error.auto.h common/error.h common/log.h common/logLevel.h common/memContext.h common/stackTrace.h common/type/buffer.h common/type/convert.h common/type/string.h
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(CMAKE) -c command/backup/common.c -o command/backup/common.o
 
+command/backup/pageChecksum.o: command/backup/pageChecksum.c build.auto.h command/backup/pageChecksum.h common/assert.h common/debug.h common/error.auto.h common/error.h common/io/filter/filter.h common/io/filter/filter.intern.h common/log.h common/logLevel.h common/macro.h common/memContext.h common/object.h common/stackTrace.h common/type/buffer.h common/type/convert.h common/type/keyValue.h common/type/string.h common/type/variant.h common/type/variantList.h postgres/pageChecksum.h
+	$(CC) $(CPPFLAGS) $(CFLAGS) $(CMAKE) -c command/backup/pageChecksum.c -o command/backup/pageChecksum.o
+
 command/command.o: command/command.c build.auto.h common/assert.h common/debug.h common/error.auto.h common/error.h common/io/filter/filter.h common/io/filter/group.h common/io/http/client.h common/io/http/header.h common/io/http/query.h common/io/read.h common/io/tls/client.h common/io/write.h common/lock.h common/log.h common/logLevel.h common/memContext.h common/stackTrace.h common/time.h common/type/buffer.h common/type/convert.h common/type/keyValue.h common/type/string.h common/type/stringList.h common/type/variant.h common/type/variantList.h config/config.auto.h config/config.h config/define.auto.h config/define.h version.h
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(CMAKE) -c command/command.c -o command/command.o
 
@@ -463,7 +466,7 @@ postgres/interface/v100.o: postgres/interface/v100.c build.auto.h common/assert.
 postgres/interface/v110.o: postgres/interface/v110.c build.auto.h common/assert.h common/debug.h common/error.auto.h common/error.h common/logLevel.h common/memContext.h common/stackTrace.h common/type/buffer.h common/type/convert.h common/type/string.h postgres/interface.h postgres/interface/version.auto.h postgres/interface/version.h postgres/interface/version.intern.h postgres/version.h
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(CMAKE) -c postgres/interface/v110.c -o postgres/interface/v110.o
 
-postgres/pageChecksum.o: postgres/pageChecksum.c build.auto.h common/assert.h common/debug.h common/error.auto.h common/error.h common/log.h common/logLevel.h common/stackTrace.h common/type/convert.h postgres/pageChecksum.h
+postgres/pageChecksum.o: postgres/pageChecksum.c build.auto.h common/assert.h common/debug.h common/error.auto.h common/error.h common/log.h common/logLevel.h common/memContext.h common/stackTrace.h common/type/buffer.h common/type/convert.h common/type/string.h postgres/interface.h postgres/pageChecksum.h
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(CMAKE) @COPTIMIZE_PAGE_CHECKSUM@ -c postgres/pageChecksum.c -o postgres/pageChecksum.o
 
 protocol/client.o: protocol/client.c build.auto.h common/assert.h common/debug.h common/error.auto.h common/error.h common/io/filter/filter.h common/io/filter/group.h common/io/read.h common/io/write.h common/log.h common/logLevel.h common/macro.h common/memContext.h common/object.h common/stackTrace.h common/time.h common/type/buffer.h common/type/convert.h common/type/json.h common/type/keyValue.h common/type/string.h common/type/variant.h common/type/variantList.h protocol/client.h protocol/command.h version.h

src/command/backup/pageChecksum.c

@@ -0,0 +1,237 @@
+/***********************************************************************************************************************************
+Page Checksum Filter
+***********************************************************************************************************************************/
+#include "build.auto.h"
+
+#include "common/debug.h"
+#include "common/io/filter/filter.intern.h"
+#include "command/backup/pageChecksum.h"
+#include "common/log.h"
+#include "common/memContext.h"
+#include "common/object.h"
+#include "postgres/pageChecksum.h"
+
+/***********************************************************************************************************************************
+Filter type constant
+***********************************************************************************************************************************/
+STRING_EXTERN(PAGE_CHECKSUM_FILTER_TYPE_STR,                        PAGE_CHECKSUM_FILTER_TYPE);
+
+/***********************************************************************************************************************************
+Object type
+***********************************************************************************************************************************/
+typedef struct PageChecksum
+{
+    MemContext *memContext;                                         // Mem context of filter
+
+    unsigned int pageNoOffset;                                      // Page number offset for subsequent segments
+    size_t pageSize;                                                // Page size
+    uint64_t lsnLimit;                                              // Lower limit of pages that could be torn
+
+    bool valid;                                                     // Is the relation structure valid?
+    bool align;                                                     // Is the relation alignment valid?
+    VariantList *error;                                             // List of checksum errors
+
+    unsigned int errorMin;                                          // Current min error page
+    unsigned int errorMax;                                          // Current max error page
+} PageChecksum;
+
+/***********************************************************************************************************************************
+Macros for function logging
+***********************************************************************************************************************************/
+String *
+pageChecksumToLog(const PageChecksum *this)
+{
+    return strNewFmt("{valid: %s, align: %s}", cvtBoolToConstZ(this->valid), cvtBoolToConstZ(this->align));
+}
+
+#define FUNCTION_LOG_PAGE_CHECKSUM_TYPE                                                                                            \
+    PageChecksum *
+#define FUNCTION_LOG_PAGE_CHECKSUM_FORMAT(value, buffer, bufferSize)                                                               \
+    FUNCTION_LOG_STRING_OBJECT_FORMAT(value, pageChecksumToLog, buffer, bufferSize)
+
+/***********************************************************************************************************************************
+Count bytes in the input
+***********************************************************************************************************************************/
+static void
+pageChecksumProcess(THIS_VOID, const Buffer *input)
+{
+    THIS(PageChecksum);
+
+    FUNCTION_LOG_BEGIN(logLevelTrace);
+        FUNCTION_LOG_PARAM(PAGE_CHECKSUM, this);
+        FUNCTION_LOG_PARAM(BUFFER, input);
+    FUNCTION_LOG_END();
+
+    ASSERT(this != NULL);
+    ASSERT(input != NULL);
+
+    // Calculate total pages in the buffer
+    unsigned int pageTotal = (unsigned int)(bufUsed(input) / this->pageSize);
+
+    // If there is a partial page make sure there is enough of it to validate the checksum
+    unsigned int pageRemainder = (unsigned int)(bufUsed(input) % this->pageSize);
+
+    if (pageRemainder != 0)
+    {
+        // Misaligned blocks, if any, should only be at the end of the file
+        if (!this->align)
+            THROW(AssertError, "should not be possible to see two misaligned pages in a row");
+
+        // Mark this buffer as misaligned in case we see another one
+        this->align = false;
+
+        // If there at least 512 bytes then we'll treat this as a partial write (modern file systems will have at least 4096)
+        if (pageRemainder >= 512)
+        {
+            pageTotal++;
+        }
+        // Else this appears to be a corrupted file and we'll stop doing page checksums
+        else
+            this->valid = false;
+    }
+
+    // Verify the checksums of complete pages in the buffer
+    if (this->valid)
+    {
+        for (unsigned int pageIdx = 0; pageIdx < pageTotal; pageIdx++)
+        {
+            unsigned char *pagePtr = bufPtr(input) + (pageIdx * this->pageSize);
+            unsigned int pageNo = this->pageNoOffset + pageIdx;
+            size_t pageSize = this->align || pageIdx < pageTotal - 1 ? this->pageSize : pageRemainder;
+
+            if (!pageChecksumTest(
+                    pagePtr, pageNo, (unsigned int)pageSize, (unsigned int)(this->lsnLimit >> 32),
+                    (unsigned int)(this->lsnLimit & 0xFFFFFFFF)))
+            {
+                MEM_CONTEXT_BEGIN(this->memContext)
+                {
+                    // Create the error list if it does not exist yet
+                    if (this->error == NULL)
+                        this->error = varLstNew();
+
+                    // Add page number and lsn to the error list
+                    VariantList *pair = varLstNew();
+                    varLstAdd(pair, varNewUInt(pageNo));
+                    varLstAdd(pair, varNewUInt64(pageLsn(pagePtr)));
+                    varLstAdd(this->error, varNewVarLst(pair));
+                }
+                MEM_CONTEXT_END();
+            }
+        }
+
+        this->pageNoOffset += pageTotal;
+    }
+
+    FUNCTION_LOG_RETURN_VOID();
+}
+
+/***********************************************************************************************************************************
+Return filter result
+***********************************************************************************************************************************/
+static Variant *
+pageChecksumResult(THIS_VOID)
+{
+    THIS(PageChecksum);
+
+    FUNCTION_LOG_BEGIN(logLevelTrace);
+        FUNCTION_LOG_PARAM(PAGE_CHECKSUM, this);
+    FUNCTION_LOG_END();
+
+    ASSERT(this != NULL);
+
+    KeyValue *result = kvNew();
+
+    if (this->error != NULL)
+    {
+        VariantList *errorList = varLstNew();
+        unsigned int errorIdx = 0;
+
+        // Convert the full list to an abbreviated list that the Perl code can understand.  In the future we want to return the
+        // entire list so pages can be verified in the WAL.
+        do
+        {
+            unsigned int pageId = varUInt(varLstGet(varVarLst(varLstGet(this->error, errorIdx)), 0));
+
+            if (errorIdx == varLstSize(this->error) - 1)
+            {
+                varLstAdd(errorList, varNewUInt(pageId));
+                errorIdx++;
+            }
+            else
+            {
+                unsigned int pageIdNext = varUInt(varLstGet(varVarLst(varLstGet(this->error, errorIdx + 1)), 0));
+
+                if (pageIdNext > pageId + 1)
+                {
+                    varLstAdd(errorList, varNewUInt(pageId));
+                    errorIdx++;
+                }
+                else
+                {
+                    unsigned int pageIdLast = pageIdNext;
+                    errorIdx++;
+
+                    while (errorIdx < varLstSize(this->error) - 1)
+                    {
+                        pageIdNext = varUInt(varLstGet(varVarLst(varLstGet(this->error, errorIdx + 1)), 0));
+
+                        if (pageIdNext > pageIdLast + 1)
+                            break;
+
+                        pageIdLast = pageIdNext;
+                        errorIdx++;
+                    }
+
+                    VariantList *errorListSub = varLstNew();
+                    varLstAdd(errorListSub, varNewUInt(pageId));
+                    varLstAdd(errorListSub, varNewUInt(pageIdLast));
+                    varLstAdd(errorList, varNewVarLst(errorListSub));
+                    errorIdx++;
+                }
+            }
+        }
+        while (errorIdx < varLstSize(this->error));
+
+        this->valid = false;
+        kvPut(result, varNewStrZ("error"), varNewVarLst(errorList));
+    }
+
+    kvPut(result, varNewStrZ("valid"), varNewBool(this->valid));
+    kvPut(result, varNewStrZ("align"), varNewBool(this->align));
+
+    FUNCTION_LOG_RETURN(VARIANT, varNewKv(result));
+}
+
+/***********************************************************************************************************************************
+New object
+***********************************************************************************************************************************/
+IoFilter *
+pageChecksumNew(unsigned int segmentNo, unsigned int segmentPageTotal, size_t pageSize, uint64_t lsnLimit)
+{
+    FUNCTION_LOG_BEGIN(logLevelTrace);
+        FUNCTION_LOG_PARAM(UINT, segmentNo);
+        FUNCTION_LOG_PARAM(UINT, segmentPageTotal);
+        FUNCTION_LOG_PARAM(SIZE, pageSize);
+        FUNCTION_LOG_PARAM(UINT64, lsnLimit);
+    FUNCTION_LOG_END();
+
+    IoFilter *this = NULL;
+
+    MEM_CONTEXT_NEW_BEGIN("PageChecksum")
+    {
+        PageChecksum *driver = memNew(sizeof(PageChecksum));
+        driver->memContext = memContextCurrent();
+
+        driver->pageNoOffset = segmentNo * segmentPageTotal;
+        driver->pageSize = pageSize;
+        driver->lsnLimit = lsnLimit;
+
+        driver->valid = true;
+        driver->align = true;
+
+        this = ioFilterNewP(PAGE_CHECKSUM_FILTER_TYPE_STR, driver, .in = pageChecksumProcess, .result = pageChecksumResult);
+    }
+    MEM_CONTEXT_NEW_END();
+
+    FUNCTION_LOG_RETURN(IO_FILTER, this);
+}

src/command/backup/pageChecksum.h

@@ -0,0 +1,22 @@
+/***********************************************************************************************************************************
+Page Checksum Filter
+
+Check all pages in a PostgreSQL relation to ensure the checksums are valid.
+***********************************************************************************************************************************/
+#ifndef COMMAND_BACKUP_PAGE_CHECKSUM_H
+#define COMMAND_BACKUP_PAGE_CHECKSUM_H
+
+#include "common/io/filter/filter.h"
+
+/***********************************************************************************************************************************
+Filter type constant
+***********************************************************************************************************************************/
+#define PAGE_CHECKSUM_FILTER_TYPE                                   "pageChecksum"
+    STRING_DECLARE(PAGE_CHECKSUM_FILTER_TYPE_STR);
+
+/***********************************************************************************************************************************
+Constructor
+***********************************************************************************************************************************/
+IoFilter *pageChecksumNew(unsigned int segmentNo, unsigned int segmentPageTotal, size_t pageSize, uint64_t lsnLimit);
+
+#endif

src/postgres/interface.c

@@ -14,13 +14,6 @@ PostgreSQL Interface
 #include "postgres/version.h"
 #include "storage/helper.h"
 
-/***********************************************************************************************************************************
-Define default page size
-
-Page size can only be changed at compile time and is not known to be well-tested, so only the default page size is supported.
-***********************************************************************************************************************************/
-#define PG_PAGE_SIZE_DEFAULT                                        ((unsigned int)(8 * 1024))
-
 /***********************************************************************************************************************************
 Define default wal segment size
 

src/postgres/interface.h

@@ -17,6 +17,21 @@ Defines for various Postgres paths and files
 #define PG_PATH_ARCHIVE_STATUS                                      "archive_status"
 #define PG_PATH_GLOBAL                                              "global"
 
+/***********************************************************************************************************************************
+Define default page size
+
+Page size can only be changed at compile time and is not known to be well-tested, so only the default page size is supported.
+***********************************************************************************************************************************/
+#define PG_PAGE_SIZE_DEFAULT                                        ((unsigned int)(8 * 1024))
+
+/***********************************************************************************************************************************
+Define default segment size and pages per segment
+
+Segment size can only be changed at compile time and is not known to be well-tested, so only the default segment size is supported.
+***********************************************************************************************************************************/
+#define PG_SEGMENT_SIZE_DEFAULT                                     ((unsigned int)(1 * 1024 * 1024 * 1024))
+#define PG_SEGMENT_PAGE_DEFAULT                                     (PG_SEGMENT_SIZE_DEFAULT / PG_PAGE_SIZE_DEFAULT)
+
 /***********************************************************************************************************************************
 PostgreSQL Control File Info
 ***********************************************************************************************************************************/

src/postgres/pageChecksum.c

@@ -69,6 +69,7 @@ minimize register spilling. For less sophisticated compilers it might be benefic
 #include "common/debug.h"
 #include "common/error.h"
 #include "common/log.h"
+#include "postgres/interface.h"
 #include "postgres/pageChecksum.h"
 
 /***********************************************************************************************************************************
@@ -203,6 +204,20 @@ pageChecksum(const unsigned char *page, unsigned int blockNo, unsigned int pageS
     FUNCTION_TEST_RETURN((uint16_t)(checksum % 65535 + 1));
 }
 
+/***********************************************************************************************************************************
+Return the lsn for a page
+***********************************************************************************************************************************/
+uint64_t
+pageLsn(const unsigned char *page)
+{
+    FUNCTION_TEST_BEGIN();
+        FUNCTION_TEST_PARAM_P(UCHARDATA, page);
+    FUNCTION_TEST_END();
+
+    // Reduce to a uint16 with an offset of one. That avoids checksums of zero, which seems like a good idea.
+    FUNCTION_TEST_RETURN((uint64_t)((PageHeader)page)->pd_lsn.walid << 32 | ((PageHeader)page)->pd_lsn.xrecoff);
+}
+
 /***********************************************************************************************************************************
 pageChecksumTest - test if checksum is valid for a single page
 ***********************************************************************************************************************************/
@@ -225,8 +240,8 @@ pageChecksumTest(
         ((PageHeader)page)->pd_upper == 0 ||
         // LSN is after the backup started so checksum is not tested because pages may be torn
         (((PageHeader)page)->pd_lsn.walid >= ignoreWalId && ((PageHeader)page)->pd_lsn.xrecoff >= ignoreWalOffset) ||
-        // Checksum is valid
+        // Checksum is valid if a full page
-        ((PageHeader)page)->pd_checksum == pageChecksum(page, blockNo, pageSize));
+        (pageSize == PG_PAGE_SIZE_DEFAULT && ((PageHeader)page)->pd_checksum == pageChecksum(page, blockNo, pageSize)));
 }
 
 /***********************************************************************************************************************************

src/postgres/pageChecksum.h

@@ -10,6 +10,7 @@ Checksum Implementation for Data Pages
 Functions
 ***********************************************************************************************************************************/
 uint16_t pageChecksum(const unsigned char *page, unsigned int blockNo, unsigned int pageSize);
+uint64_t pageLsn(const unsigned char *page);
 bool pageChecksumTest(
     const unsigned char *page, unsigned int blockNo, unsigned int pageSize, uint32_t ignoreWalId, uint32_t ignoreWalOffset);
 bool pageChecksumBufferTest(

test/define.yaml

@@ -671,10 +671,11 @@ unit:
 
       # ----------------------------------------------------------------------------------------------------------------------------
       - name: backup-common
-        total: 1
+        total: 2
 
         coverage:
           command/backup/common: full
+          command/backup/pageChecksum: full
 
       # ----------------------------------------------------------------------------------------------------------------------------
       - name: command

test/src/module/command/backupCommonTest.c

@@ -2,9 +2,30 @@
 Test Common Functions and Definitions for Backup and Expire Commands
 ***********************************************************************************************************************************/
 #include "common/harnessConfig.h"
+#include "common/io/bufferWrite.h"
+#include "common/regExp.h"
+#include "common/type/json.h"
+#include "postgres/interface.h"
 #include "storage/posix/storage.h"
 
-#include <common/regExp.h>
+/***********************************************************************************************************************************
+Need these structures to mock up test data
+***********************************************************************************************************************************/
+typedef struct
+{
+    uint32_t walid;                                                 // high bits
+    uint32_t xrecoff;                                               // low bits
+} PageWalRecPtr;
+
+typedef struct PageHeaderData
+{
+    // LSN is member of *any* block, not only page-organized ones
+    PageWalRecPtr pd_lsn;                                           // Lsn for last change to this page
+    uint16_t pd_checksum;                                           // checksum
+    uint16_t pd_flags;                                              // flag bits, see below
+    uint16_t pd_lower;                                              // offset to start of free space
+    uint16_t pd_upper;                                              // offset to end of free space
+} PageHeaderData;
 
 /***********************************************************************************************************************************
 Test Run
@@ -106,5 +127,132 @@ testRun(void)
         TEST_RESULT_BOOL(regExpMatchOne(filter, full), false, "   does not match full");
     }
 
+    // *****************************************************************************************************************************
+    if (testBegin("PageChecksum"))
+    {
+        TEST_RESULT_UINT(PG_SEGMENT_PAGE_DEFAULT, 131072, "check pages per segment");
+
+        // Test pages with all zeros (these are considered valid)
+        // -------------------------------------------------------------------------------------------------------------------------
+        Buffer *buffer = bufNew(PG_PAGE_SIZE_DEFAULT * 3);
+        Buffer *bufferOut = bufNew(0);
+        bufUsedSet(buffer, bufSize(buffer));
+        memset(bufPtr(buffer), 0, bufSize(buffer));
+
+        IoWrite *write = ioWriteFilterGroupSet(
+            ioBufferWriteNew(bufferOut),
+            ioFilterGroupAdd(ioFilterGroupNew(), pageChecksumNew(0, PG_SEGMENT_PAGE_DEFAULT, PG_PAGE_SIZE_DEFAULT, 0)));
+        ioWriteOpen(write);
+        ioWrite(write, buffer);
+        ioWriteClose(write);
+
+        TEST_RESULT_STR(
+            strPtr(jsonFromVar(ioFilterGroupResult(ioWriteFilterGroup(write), PAGE_CHECKSUM_FILTER_TYPE_STR), 0)),
+            "{\"align\":true,\"valid\":true}", "all zero pages");
+
+        // Single checksum error
+        // -------------------------------------------------------------------------------------------------------------------------
+        buffer = bufNew(PG_PAGE_SIZE_DEFAULT * 1);
+        bufUsedSet(buffer, bufSize(buffer));
+        memset(bufPtr(buffer), 0, bufSize(buffer));
+
+        // Page 0 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_lsn.walid = 0xF0F0F0F0;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_lsn.xrecoff = 0xF0F0F0F0;
+
+        write = ioWriteFilterGroupSet(
+            ioBufferWriteNew(bufferOut),
+            ioFilterGroupAdd(
+                ioFilterGroupNew(), pageChecksumNew(0, PG_SEGMENT_PAGE_DEFAULT, PG_PAGE_SIZE_DEFAULT, 0xFACEFACE00000000)));
+        ioWriteOpen(write);
+        ioWrite(write, buffer);
+        ioWriteClose(write);
+
+        TEST_RESULT_STR(
+            strPtr(jsonFromVar(ioFilterGroupResult(ioWriteFilterGroup(write), PAGE_CHECKSUM_FILTER_TYPE_STR), 0)),
+            "{\"align\":true,\"error\":[0],\"valid\":false}", "single checksum error");
+
+        // Various checksum errors some of which will be skipped because of the LSN
+        // -------------------------------------------------------------------------------------------------------------------------
+        buffer = bufNew(PG_PAGE_SIZE_DEFAULT * 8 - (PG_PAGE_SIZE_DEFAULT - 512));
+        bufUsedSet(buffer, bufSize(buffer));
+        memset(bufPtr(buffer), 0, bufSize(buffer));
+
+        // Page 0 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_lsn.walid = 0xF0F0F0F0;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x00)))->pd_lsn.xrecoff = 0xF0F0F0F0;
+
+        // Page 1 has bogus checksum but lsn above the limit
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x01)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x01)))->pd_lsn.walid = 0xFACEFACE;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x01)))->pd_lsn.xrecoff = 0x00000000;
+
+        // Page 2 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x02)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x02)))->pd_lsn.xrecoff = 0x2;
+
+        // Page 3 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x03)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x03)))->pd_lsn.xrecoff = 0x3;
+
+        // Page 4 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x04)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x04)))->pd_lsn.xrecoff = 0x4;
+
+        // Page 6 has bogus checksum
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x06)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x06)))->pd_lsn.xrecoff = 0x6;
+
+        // Page 7 has bogus checksum (and is misaligned but large enough to test)
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x07)))->pd_upper = 0x01;
+        ((PageHeaderData *)(bufPtr(buffer) + (PG_PAGE_SIZE_DEFAULT * 0x07)))->pd_lsn.xrecoff = 0x7;
+
+        write = ioWriteFilterGroupSet(
+            ioBufferWriteNew(bufferOut),
+            ioFilterGroupAdd(
+                ioFilterGroupNew(), pageChecksumNew(0, PG_SEGMENT_PAGE_DEFAULT, PG_PAGE_SIZE_DEFAULT, 0xFACEFACE00000000)));
+        ioWriteOpen(write);
+        ioWrite(write, buffer);
+        ioWriteClose(write);
+
+        TEST_RESULT_STR(
+            strPtr(jsonFromVar(ioFilterGroupResult(ioWriteFilterGroup(write), PAGE_CHECKSUM_FILTER_TYPE_STR), 0)),
+            "{\"align\":false,\"error\":[0,[2,4],[6,7]],\"valid\":false}", "various checksum errors");
+
+        // Impossibly misaligned page
+        // -------------------------------------------------------------------------------------------------------------------------
+        buffer = bufNew(256);
+        bufUsedSet(buffer, bufSize(buffer));
+        memset(bufPtr(buffer), 0, bufSize(buffer));
+
+        write = ioWriteFilterGroupSet(
+            ioBufferWriteNew(bufferOut),
+            ioFilterGroupAdd(
+                ioFilterGroupNew(), pageChecksumNew(0, PG_SEGMENT_PAGE_DEFAULT, PG_PAGE_SIZE_DEFAULT, 0xFACEFACE00000000)));
+        ioWriteOpen(write);
+        ioWrite(write, buffer);
+        ioWriteClose(write);
+
+        TEST_RESULT_STR(
+            strPtr(jsonFromVar(ioFilterGroupResult(ioWriteFilterGroup(write), PAGE_CHECKSUM_FILTER_TYPE_STR), 0)),
+            "{\"align\":false,\"valid\":false}", "misalignment");
+
+        // Two misaligned buffers in a row
+        // -------------------------------------------------------------------------------------------------------------------------
+        buffer = bufNew(513);
+        bufUsedSet(buffer, bufSize(buffer));
+        memset(bufPtr(buffer), 0, bufSize(buffer));
+
+        write = ioWriteFilterGroupSet(
+            ioBufferWriteNew(bufferOut),
+            ioFilterGroupAdd(
+                ioFilterGroupNew(), pageChecksumNew(0, PG_SEGMENT_PAGE_DEFAULT, PG_PAGE_SIZE_DEFAULT, 0xFACEFACE00000000)));
+        ioWriteOpen(write);
+        ioWrite(write, buffer);
+        TEST_ERROR(ioWrite(write, buffer), AssertError, "should not be possible to see two misaligned pages in a row");
+    }
+
     FUNCTION_HARNESS_RESULT_VOID();
 }

test/src/module/postgres/pageChecksumTest.c

@@ -42,7 +42,7 @@ testRun(void)
     }
 
     // *****************************************************************************************************************************
-    if (testBegin("pageChecksumTest()"))
+    if (testBegin("pageChecksumTest() and pageLsn()"))
     {
         // Zero the pages
         memset(testPage(0), 0, TEST_PAGE_TOTAL * TEST_PAGE_SIZE);
@@ -51,8 +51,15 @@ testRun(void)
         TEST_RESULT_BOOL(pageChecksumTest(testPage(0), 0, TEST_PAGE_SIZE, 0, 0), true, "pd_upper is 0, block 0");
         TEST_RESULT_BOOL(pageChecksumTest(testPage(1), 999, TEST_PAGE_SIZE, 0, 0), true, "pd_upper is 0, block 999");
 
+        // Partial pages are always invalid
+        ((PageHeader)testPage(0))->pd_upper = 0x00FF;
+        ((PageHeader)testPage(0))->pd_checksum = pageChecksum(testPage(0), 0, TEST_PAGE_SIZE);
+        TEST_RESULT_BOOL(pageChecksumTest(testPage(0), 0, TEST_PAGE_SIZE, 1, 1), true, "valid page");
+        TEST_RESULT_BOOL(pageChecksumTest(testPage(0), 0, TEST_PAGE_SIZE / 2, 1, 1), false, "invalid partial page");
+
         // Update pd_upper and check for failure no matter the block no
         ((PageHeader)testPage(0))->pd_upper = 0x00FF;
+        ((PageHeader)testPage(0))->pd_checksum = 0;
         TEST_RESULT_BOOL(pageChecksumTest(testPage(0), 0, TEST_PAGE_SIZE, 0xFFFF, 0xFFFF), false, "badchecksum, page 0");
         TEST_RESULT_BOOL(
             pageChecksumTest(testPage(0), 9999, TEST_PAGE_SIZE, 0xFFFF, 0xFFFF), false, "badchecksum, page 9999");
@@ -61,6 +68,8 @@ testRun(void)
         ((PageHeader)testPage(0))->pd_lsn.walid = 0x8888;
         ((PageHeader)testPage(0))->pd_lsn.xrecoff = 0x8888;
 
+        TEST_RESULT_UINT(pageLsn(testPage(0)), 0x0000888800008888, "check page lsn");
+
         TEST_RESULT_BOOL(
             pageChecksumTest(testPage(0), 0, TEST_PAGE_SIZE, 0x8888, 0x8888), true, "bad checksum past ignore limit");
         TEST_RESULT_BOOL(