Re: astreamer_lz4: fix bug of output pointer advancement in decompressor - Mailing list pgsql-hackers

I wrote:
> I suspect whoever wrote this thought pg_log_error is equivalent
> to elog(ERROR), but it's not; it just prints a message.  It seems
> highly unlikely to me that continuing onwards will result in a
> good outcome.  I'm a bit inclined to s/pg_log_error/pg_fatal/
> throughout these files, at least in places where there's no
> visible effort to handle the error.

After looking through fe_utils, pg_dump, pg_basebackup, and
pg_verifybackup, I found the attached places that seem to
need cleanup.  There are a couple other places where we
are not treating failures as fatal, but those seem intentional,
eg not fatal'ing on close() failure for an input file.

            regards, tom lane

diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index b72bad130ad..0aa519fbb67 100644
--- 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 e8d62f754ca..2e080c37a58 100644
--- 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 e196fcc81e5..2bc32b42879 100644
--- 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 2bf5c57b902..f26abcfd0fa 100644
--- 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));
     }
 }