From f4a4af299bee93ab7dde599ccdb35becc1c2afad Mon Sep 17 00:00:00 2001 From: David Steele Date: Wed, 27 Dec 2023 12:49:47 -0300 Subject: [PATCH] Fix short read in block incremental restore. During restore it is possible to read all the blocks out of a compressed super block without reading all the input. This is because the compression format may have some trailing bytes that are not required for decompression but are required to indicate that data has ended. If a buffer aligned with the compressed data in a certain way, these last bytes might not be read. Explicitly read out any final bytes at the end of each super block to handle this case. This should always result in no additional data out and we check for that, but it does move the read position to the beginning of the next compressed super block so decompression can begin without error. --- doc/xml/release/2024/2.50.xml | 17 ++++++++ doc/xml/release/contributor.xml | 10 +++++ src/command/restore/blockDelta.c | 5 +++ src/common/io/read.c | 36 +++++++++++++++++ src/common/io/read.h | 12 ++++++ test/define.yaml | 2 +- test/src/module/command/restoreTest.c | 56 +++++++++++++++++++++++++++ test/src/module/common/ioTest.c | 25 ++++++++++++ 8 files changed, 162 insertions(+), 1 deletion(-) diff --git a/doc/xml/release/2024/2.50.xml b/doc/xml/release/2024/2.50.xml index cc9b5d2dd..3ec8b96d6 100644 --- a/doc/xml/release/2024/2.50.xml +++ b/doc/xml/release/2024/2.50.xml @@ -1,6 +1,23 @@ + + + + + + + + + + + + + + +

Fix short read in block incremental restore.

+ + diff --git a/doc/xml/release/contributor.xml b/doc/xml/release/contributor.xml index df7daa147..75773f022 100644 --- a/doc/xml/release/contributor.xml +++ b/doc/xml/release/contributor.xml @@ -25,6 +25,11 @@ Flamacue + + Adol Rodriguez + adolsalamanca + + Adrian Vondendriesch disco-stu @@ -120,6 +125,11 @@ bradnicholson + + Brent Graveland + graveland + + Brian Almeida thunderkeys diff --git a/src/command/restore/blockDelta.c b/src/command/restore/blockDelta.c index f7e2a5848..7ab3b7c50 100644 --- a/src/command/restore/blockDelta.c +++ b/src/command/restore/blockDelta.c @@ -293,6 +293,11 @@ blockDeltaNext(BlockDelta *const this, const BlockDeltaRead *const readDelta, Io if (result != NULL) break; + // Check that no bytes remain to be written. It is possible that some bytes remain in the super block, however, since we may + // have gotten all the bytes we needed but just missed reading something important, e.g. an end of file marker. If we do not + // read the remaining bytes then the next read will start too early. + ioReadFlushP(this->limitRead, .errorOnBytes = true); + this->superBlockData = NULL; this->superBlockIdx++; } diff --git a/src/common/io/read.c b/src/common/io/read.c index adc3a5828..0cb9b3620 100644 --- a/src/common/io/read.c +++ b/src/common/io/read.c @@ -427,6 +427,42 @@ ioReadReady(IoRead *this, IoReadReadyParam param) FUNCTION_LOG_RETURN(BOOL, result); } +/**********************************************************************************************************************************/ +FN_EXTERN uint64_t +ioReadFlush(IoRead *const this, const IoReadFlushParam param) +{ + FUNCTION_LOG_BEGIN(logLevelTrace); + FUNCTION_LOG_PARAM(IO_READ, this); + FUNCTION_LOG_PARAM(BOOL, param.errorOnBytes); + FUNCTION_LOG_END(); + + ASSERT(this != NULL); + ASSERT(this->pub.opened && !this->pub.closed); + + // Flush remaining data + uint64_t result = 0; + + if (!ioReadEof(this)) + { + Buffer *const buffer = bufNew(ioBufferSize()); + + do + { + result += ioRead(this, buffer); + bufUsedZero(buffer); + } + while (!ioReadEof(this)); + + bufFree(buffer); + } + + // Error when bytes found and error requested + if (result != 0 && param.errorOnBytes) + THROW_FMT(FileReadError, "expected EOF but flushed %" PRIu64 " byte(s)", result); + + FUNCTION_LOG_RETURN(UINT64, result); +} + /**********************************************************************************************************************************/ FN_EXTERN void ioReadClose(IoRead *this) diff --git a/src/common/io/read.h b/src/common/io/read.h index 0826f71f2..bec653578 100644 --- a/src/common/io/read.h +++ b/src/common/io/read.h @@ -85,6 +85,18 @@ typedef struct IoReadReadyParam FN_EXTERN bool ioReadReady(IoRead *this, IoReadReadyParam param); +// Flush all remaining bytes and return bytes flushed. Optionally error when bytes are flushed. +typedef struct IoReadFlushParam +{ + VAR_PARAM_HEADER; + bool errorOnBytes; // Error when any bytes are found during flush +} IoReadFlushParam; + +#define ioReadFlushP(this, ...) \ + ioReadFlush(this, (IoReadFlushParam){VAR_PARAM_INIT, __VA_ARGS__}) + +FN_EXTERN uint64_t ioReadFlush(IoRead *this, IoReadFlushParam param); + // Close the IO FN_EXTERN void ioReadClose(IoRead *this); diff --git a/test/define.yaml b/test/define.yaml index 70a360e1d..fd90a9324 100644 --- a/test/define.yaml +++ b/test/define.yaml @@ -907,7 +907,7 @@ unit: # ---------------------------------------------------------------------------------------------------------------------------- - name: restore - total: 14 + total: 15 coverage: - command/restore/blockChecksum diff --git a/test/src/module/command/restoreTest.c b/test/src/module/command/restoreTest.c index 6b9ddff48..52910a7c4 100644 --- a/test/src/module/command/restoreTest.c +++ b/test/src/module/command/restoreTest.c @@ -7,11 +7,13 @@ Test Restore Command #include "command/stanza/create.h" #include "common/compress/helper.h" #include "common/crypto/cipherBlock.h" +#include "common/io/bufferRead.h" #include "postgres/version.h" #include "storage/helper.h" #include "storage/posix/storage.h" #include "common/harnessBackup.h" +#include "common/harnessBlockIncr.h" #include "common/harnessConfig.h" #include "common/harnessInfo.h" #include "common/harnessManifest.h" @@ -206,6 +208,60 @@ testRun(void) ioWriteFree(write); } + // ***************************************************************************************************************************** + if (testBegin("BlockDelta")) + { + // ------------------------------------------------------------------------------------------------------------------------- + TEST_TITLE("super blocks are read until the end"); + + // Write block incremental with multiple super blocks + const Buffer *source = BUFSTRZ("123456789ABC"); + Buffer *destination = bufNew(256); + IoWrite *write = ioBufferWriteNew(destination); + + ioFilterGroupAdd( + ioWriteFilterGroup(write), blockIncrNew(6, 3, 5, 0, 0, 0, NULL, compressFilterP(compressTypeGz, 1, .raw = true), NULL)); + ioWriteOpen(write); + ioWrite(write, source); + ioWriteClose(write); + + // Extract block map + uint64_t mapSize = pckReadU64P(ioFilterGroupResultP(ioWriteFilterGroup(write), BLOCK_INCR_FILTER_TYPE)); + const BlockMap *blockMap = blockMapNewRead( + ioBufferReadNewOpen(BUF(bufPtr(destination) + (bufUsed(destination) - (size_t)mapSize), (size_t)mapSize)), 3, 5); + + // Perform block delta + BlockDelta *blockDelta = blockDeltaNew(blockMap, 3, 5, NULL, cipherTypeNone, NULL, compressTypeGz); + const BlockDeltaRead *blockDeltaRead = blockDeltaReadGet(blockDelta, 0); + IoRead *read = ioBufferReadNewOpen(destination); + + // Set buffer size to one to make sure super blocks are properly flushed + size_t bufferSizeOld = ioBufferSize(); + ioBufferSizeSet(1); + + // Read all blocks from multiple super blocks + TEST_RESULT_STR_Z(strNewBuf(blockDeltaNext(blockDelta, blockDeltaRead, read)->block), "123", "read block"); + TEST_RESULT_STR_Z(strNewBuf(blockDeltaNext(blockDelta, blockDeltaRead, read)->block), "456", "read block"); + TEST_RESULT_STR_Z(strNewBuf(blockDeltaNext(blockDelta, blockDeltaRead, read)->block), "789", "read block"); + TEST_RESULT_STR_Z(strNewBuf(blockDeltaNext(blockDelta, blockDeltaRead, read)->block), "ABC", "read block"); + TEST_RESULT_PTR(blockDeltaNext(blockDelta, blockDeltaRead, read), NULL, "no more blocks"); + + // Reset buffer size + ioBufferSizeSet(bufferSizeOld); + + // Now test the block map to show that we are reading in the right position after multiple super blocks + TEST_RESULT_STR_Z( + hrnBlockDeltaRender(blockMapNewRead(read, 3, 5), 3, 5), + "read {reference: 0, bundleId: 0, offset: 0, size: 28}\n" + " super block {max: 6, size: 14}\n" + " block {no: 0, offset: 0}\n" + " block {no: 1, offset: 3}\n" + " super block {max: 6, size: 14}\n" + " block {no: 0, offset: 6}\n" + " block {no: 1, offset: 9}\n", + "check delta"); + } + // ***************************************************************************************************************************** if (testBegin("restoreFile()")) { diff --git a/test/src/module/common/ioTest.c b/test/src/module/common/ioTest.c index 73f69ac73..656dc8db7 100644 --- a/test/src/module/common/ioTest.c +++ b/test/src/module/common/ioTest.c @@ -296,6 +296,31 @@ testRun(void) TEST_RESULT_BOOL(ioReadOpen(bufferRead), true, " open"); TEST_RESULT_BOOL(ioReadEof(bufferRead), false, " not eof"); TEST_RESULT_UINT(ioRead(bufferRead, buffer), 0, " read 0 bytes"); + TEST_RESULT_UINT(ioReadFlushP(bufferRead), 0, " flush 0 bytes"); + TEST_RESULT_BOOL(ioReadEof(bufferRead), true, " now eof"); + + // ------------------------------------------------------------------------------------------------------------------------- + TEST_TITLE("read flush bytes"); + + ioBufferSizeSet(1); + bufferOriginal = bufNewC("123", 3); + + TEST_ASSIGN(bufferRead, ioBufferReadNew(bufferOriginal), "create empty buffer read object"); + TEST_RESULT_BOOL(ioReadOpen(bufferRead), true, " open"); + TEST_RESULT_BOOL(ioReadEof(bufferRead), false, " not eof"); + TEST_RESULT_UINT(ioReadFlushP(bufferRead), 3, " flush 3 bytes"); + TEST_RESULT_BOOL(ioReadEof(bufferRead), true, " now eof"); + + // ------------------------------------------------------------------------------------------------------------------------- + TEST_TITLE("read flush bytes and error"); + + ioBufferSizeSet(1); + bufferOriginal = bufNewC("123", 3); + + TEST_ASSIGN(bufferRead, ioBufferReadNew(bufferOriginal), "create empty buffer read object"); + TEST_RESULT_BOOL(ioReadOpen(bufferRead), true, " open"); + TEST_RESULT_BOOL(ioReadEof(bufferRead), false, " not eof"); + TEST_ERROR(ioReadFlushP(bufferRead, .errorOnBytes = true), FileReadError, "expected EOF but flushed 3 byte(s)"); TEST_RESULT_BOOL(ioReadEof(bufferRead), true, " now eof"); // -------------------------------------------------------------------------------------------------------------------------