Exit after fatal errors in client-side compression code.

It looks like whoever wrote the astreamer (nee bbstreamer) code
thought that pg_log_error() is equivalent to elog(ERROR), but
it's not; it just prints a message.  So all these places tried to
continue on after a compression or decompression error return,
with the inevitable result being garbage output and possibly
cascading error messages.  We should use pg_fatal() instead.

These error conditions are probably pretty unlikely in practice,
which no doubt accounts for the lack of field complaints.

Author: Tom Lane <tgl@sss.pgh.pa.us>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://postgr.es/m/1531718.1772644615@sss.pgh.pa.us
Backpatch-through: 15

diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index b72bad130ad6d26d4736e298c3988dad6c46687e..0aa519fbb6741c621a9669df80d73793246e325d 100644 (file)
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -629,6 +629,7 @@ LZ4Stream_close(CompressFileHandle *CFH)
        size_t          required;
        size_t          status;
        int                     ret;
+       bool            success = true;
 
        fp = state->fp;
        if (state->inited)
@@ -644,6 +645,7 @@ LZ4Stream_close(CompressFileHandle *CFH)
                                {
                                        errno = (errno) ? errno : ENOSPC;
                                        pg_log_error("could not write to output file: %m");
+                                       success = false;
                                }
                                state->bufdata = 0;
                        }
@@ -656,6 +658,7 @@ LZ4Stream_close(CompressFileHandle *CFH)
                        {
                                pg_log_error("could not end compression: %s",
                                                         LZ4F_getErrorName(status));
+                               success = false;
                        }
                        else
                                state->bufdata += status;
@@ -665,19 +668,26 @@ LZ4Stream_close(CompressFileHandle *CFH)
                        {
                                errno = (errno) ? errno : ENOSPC;
                                pg_log_error("could not write to output file: %m");
+                               success = false;
                        }
 
                        status = LZ4F_freeCompressionContext(state->ctx);
                        if (LZ4F_isError(status))
+                       {
                                pg_log_error("could not end compression: %s",
                                                         LZ4F_getErrorName(status));
+                               success = false;
+                       }
                }
                else
                {
                        status = LZ4F_freeDecompressionContext(state->dtx);
                        if (LZ4F_isError(status))
+                       {
                                pg_log_error("could not end decompression: %s",
                                                         LZ4F_getErrorName(status));
+                               success = false;
+                       }
                        pg_free(state->outbuf);
                }
 
@@ -692,10 +702,10 @@ LZ4Stream_close(CompressFileHandle *CFH)
        if (ret != 0)
        {
                pg_log_error("could not close file: %m");
-               return false;
+               success = false;
        }
 
-       return true;
+       return success;
 }
 
 static bool

diff --git a/src/fe_utils/astreamer_gzip.c b/src/fe_utils/astreamer_gzip.c
index e8d62f754cadddec99583eac03803bf46edad09f..2e080c37a5854aefbde9e986c5c9b397d024b702 100644 (file)
--- a/src/fe_utils/astreamer_gzip.c
+++ b/src/fe_utils/astreamer_gzip.c
@@ -317,7 +317,7 @@ astreamer_gzip_decompressor_content(astreamer *streamer,
                res = inflate(zs, Z_NO_FLUSH);
 
                if (res == Z_STREAM_ERROR)
-                       pg_log_error("could not decompress data: %s", zs->msg);
+                       pg_fatal("could not decompress data: %s", zs->msg);
 
                mystreamer->bytes_written =
                        mystreamer->base.bbs_buffer.maxlen - zs->avail_out;

diff --git a/src/fe_utils/astreamer_lz4.c b/src/fe_utils/astreamer_lz4.c
index e196fcc81e5817ca2bb58be34bd01ae498c499de..2bc32b42879b0e58d5b9a19a53869cd0d3b871ae 100644 (file)
--- a/src/fe_utils/astreamer_lz4.c
+++ b/src/fe_utils/astreamer_lz4.c
@@ -94,8 +94,8 @@ astreamer_lz4_compressor_new(astreamer *next, pg_compress_specification *compres
 
        ctxError = LZ4F_createCompressionContext(&streamer->cctx, LZ4F_VERSION);
        if (LZ4F_isError(ctxError))
-               pg_log_error("could not create lz4 compression context: %s",
-                                        LZ4F_getErrorName(ctxError));
+               pg_fatal("could not create lz4 compression context: %s",
+                                LZ4F_getErrorName(ctxError));
 
        return &streamer->base;
 #else
@@ -139,8 +139,8 @@ astreamer_lz4_compressor_content(astreamer *streamer,
                                                                                         &mystreamer->prefs);
 
                if (LZ4F_isError(compressed_size))
-                       pg_log_error("could not write lz4 header: %s",
-                                                LZ4F_getErrorName(compressed_size));
+                       pg_fatal("could not write lz4 header: %s",
+                                        LZ4F_getErrorName(compressed_size));
 
                mystreamer->bytes_written += compressed_size;
                mystreamer->header_written = true;
@@ -188,8 +188,8 @@ astreamer_lz4_compressor_content(astreamer *streamer,
                                                                                  next_in, len, NULL);
 
        if (LZ4F_isError(compressed_size))
-               pg_log_error("could not compress data: %s",
-                                        LZ4F_getErrorName(compressed_size));
+               pg_fatal("could not compress data: %s",
+                                LZ4F_getErrorName(compressed_size));
 
        mystreamer->bytes_written += compressed_size;
 }
@@ -240,8 +240,8 @@ astreamer_lz4_compressor_finalize(astreamer *streamer)
                                                                           next_out, avail_out, NULL);
 
        if (LZ4F_isError(compressed_size))
-               pg_log_error("could not end lz4 compression: %s",
-                                        LZ4F_getErrorName(compressed_size));
+               pg_fatal("could not end lz4 compression: %s",
+                                LZ4F_getErrorName(compressed_size));
 
        mystreamer->bytes_written += compressed_size;
 
@@ -353,8 +353,8 @@ astreamer_lz4_decompressor_content(astreamer *streamer,
                                                          next_in, &read_size, NULL);
 
                if (LZ4F_isError(ret))
-                       pg_log_error("could not decompress data: %s",
-                                                LZ4F_getErrorName(ret));
+                       pg_fatal("could not decompress data: %s",
+                                        LZ4F_getErrorName(ret));
 
                /* Update input buffer based on number of bytes consumed */
                avail_in -= read_size;

diff --git a/src/fe_utils/astreamer_zstd.c b/src/fe_utils/astreamer_zstd.c
index 2bf5c57b9029a25ebf8aa83b99e5589f2a331326..f26abcfd0fa5d0bf680d4a7f9c9f67516a2aaffd 100644 (file)
--- a/src/fe_utils/astreamer_zstd.c
+++ b/src/fe_utils/astreamer_zstd.c
@@ -116,11 +116,8 @@ astreamer_zstd_compressor_new(astreamer *next, pg_compress_specification *compre
                                                                         ZSTD_c_enableLongDistanceMatching,
                                                                         compress->long_distance);
                if (ZSTD_isError(ret))
-               {
-                       pg_log_error("could not enable long-distance mode: %s",
-                                                ZSTD_getErrorName(ret));
-                       exit(1);
-               }
+                       pg_fatal("could not enable long-distance mode: %s",
+                                        ZSTD_getErrorName(ret));
        }
 
        /* Initialize the ZSTD output buffer. */
@@ -182,8 +179,8 @@ astreamer_zstd_compressor_content(astreamer *streamer,
                                                                 &inBuf, ZSTD_e_continue);
 
                if (ZSTD_isError(yet_to_flush))
-                       pg_log_error("could not compress data: %s",
-                                                ZSTD_getErrorName(yet_to_flush));
+                       pg_fatal("could not compress data: %s",
+                                        ZSTD_getErrorName(yet_to_flush));
        }
 }
 
@@ -224,8 +221,8 @@ astreamer_zstd_compressor_finalize(astreamer *streamer)
                                                                                        &in, ZSTD_e_end);
 
                if (ZSTD_isError(yet_to_flush))
-                       pg_log_error("could not compress data: %s",
-                                                ZSTD_getErrorName(yet_to_flush));
+                       pg_fatal("could not compress data: %s",
+                                        ZSTD_getErrorName(yet_to_flush));
 
        } while (yet_to_flush > 0);
 
@@ -330,8 +327,8 @@ astreamer_zstd_decompressor_content(astreamer *streamer,
                                                                        &mystreamer->zstd_outBuf, &inBuf);
 
                if (ZSTD_isError(ret))
-                       pg_log_error("could not decompress data: %s",
-                                                ZSTD_getErrorName(ret));
+                       pg_fatal("could not decompress data: %s",
+                                        ZSTD_getErrorName(ret));
        }
 }