Fix gprestore/gpbackup hanging in case the helper goes down

end_to_end/end_to_end_suite_test.go

@@ -2302,7 +2302,7 @@ LANGUAGE plpgsql NO SQL;`)
 			output, err := gprestoreCmd.CombinedOutput()
 			Expect(err).To(HaveOccurred())
 			Expect(string(output)).To(ContainSubstring(`Error loading data into table public.t1`))
-			Expect(string(output)).To(ContainSubstring(`Error loading data into table public.t2`))
+			Expect(string(output)).To(SatisfyAny(ContainSubstring(`Error loading data into table public.t2`), ContainSubstring(`Expected to restore 1000000 rows to table public.t2, but restored 0 instead`)))
 			assertArtifactsCleaned(restoreConn, "20240502095933")
 			testhelper.AssertQueryRuns(restoreConn, "DROP TABLE t0; DROP TABLE t1; DROP TABLE t2; DROP TABLE t3; DROP TABLE t4;")
 		})

end_to_end/plugin_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"os/exec"
 	path "path/filepath"
+	"time"
 
 	"github.com/greenplum-db/gp-common-go-libs/cluster"
 	"github.com/greenplum-db/gp-common-go-libs/dbconn"
@@ -14,6 +15,7 @@ import (
 	"github.com/greenplum-db/gpbackup/testutils"
 	"github.com/greenplum-db/gpbackup/utils"
 	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
 )
 
 func copyPluginToAllHosts(conn *dbconn.DBConn, pluginPath string) {
@@ -306,6 +308,42 @@ var _ = Describe("End to End plugin tests", func() {
 					Skip("This test is only needed for the most recent backup versions")
 				}
 			})
+			It("Will not hang if gpbackup and gprestore runs with single-data-file and plugin fails at init for restore", func(ctx SpecContext) {
+				pluginExecutablePath := fmt.Sprintf("%s/src/github.com/greenplum-db/gpbackup/plugins/example_plugin.bash", os.Getenv("GOPATH"))
+				copyPluginToAllHosts(backupConn, pluginExecutablePath)
+
+				testhelper.AssertQueryRuns(backupConn, "CREATE TABLE t0(a int);")
+				testhelper.AssertQueryRuns(backupConn, "INSERT INTO t0 SELECT i FROM generate_series(1, 10)i;")
+				defer testhelper.AssertQueryRuns(backupConn, "DROP TABLE t0;")
+
+				timestamp := gpbackup(gpbackupPath, backupHelperPath,
+					"--single-data-file",
+					"--plugin-config", pluginConfigPath)
+
+				backupCluster.GenerateAndExecuteCommand(
+					"Instruct plugin to fail",
+					cluster.ON_HOSTS,
+					func(contentID int) string {
+						return fmt.Sprintf("touch /tmp/GPBACKUP_PLUGIN_FAIL")
+					})
+
+				gprestoreCmd := exec.Command(gprestorePath,
+					"--timestamp", timestamp,
+					"--redirect-db", "restoredb",
+					"--plugin-config", pluginConfigPath)
+
+				_, err := gprestoreCmd.CombinedOutput()
+				Expect(err).To(HaveOccurred())
+
+				backupCluster.GenerateAndExecuteCommand(
+					"Unset plugin failure",
+					cluster.ON_HOSTS,
+					func(contentID int) string {
+						return fmt.Sprintf("rm /tmp/GPBACKUP_PLUGIN_FAIL")
+					})
+
+				assertArtifactsCleaned(restoreConn, timestamp)
+			}, SpecTimeout(time.Second*30))
 			It("runs gpbackup and gprestore with plugin, single-data-file, and no-compression", func() {
 				pluginExecutablePath := fmt.Sprintf("%s/src/github.com/greenplum-db/gpbackup/plugins/example_plugin.bash", os.Getenv("GOPATH"))
 				copyPluginToAllHosts(backupConn, pluginExecutablePath)

helper/helper.go

@@ -256,6 +256,33 @@ func DoCleanup() {
 
 	for pipeName, _ := range pipesMap {
 		log("Removing pipe %s", pipeName)
+
+		/*
+		 * In case restore helper encountered an error, it is possible that the pipe file has been already opened for
+		 * reading by another process, but nothing has been written to the pipe yet. In this case, the reader process
+		 * may hang forever after the pipe is removed. To fix this problem, the pipe file is opened and closed
+		 * to generate an EOF before it is deleted.
+		 *
+		 * Also, it is possible, that after EOF is issued (releasing the current reader process), but before the pipe is
+		 * removed, a new reader process can start reading the pipe. To avoid such situation, we close the file handle
+		 * after we have removed the file.
+		 *
+		 * Similar problem can happen with backup helper. If it finishes its work right after start, before opening the
+		 * pipe for reading, the writing side will stay hanging on the other pipe's end. So, we also open and close the
+		 * pipe to release the writing side.
+		 */
+		if *restoreAgent {
+			fileHandlePipe, err := os.OpenFile(pipeName, os.O_WRONLY|unix.O_NONBLOCK, os.ModeNamedPipe)
+			if err == nil {
+				defer fileHandlePipe.Close()
+			}
+		} else if *backupAgent {
+			fileHandlePipe, err := os.OpenFile(pipeName, os.O_RDONLY|unix.O_NONBLOCK, os.ModeNamedPipe)
+			if err == nil {
+				defer fileHandlePipe.Close()
+			}
+		}
+
 		err = deletePipe(pipeName)
 		if err != nil {
 			log("Encountered error removing pipe %s: %v", pipeName, err)

helper/restore_helper.go

@@ -126,6 +126,8 @@ func doRestoreAgent() error {
 		}
 	}
 
+	preloadCreatedPipes(oidList, *copyQueue)
+
 	if *singleDataFile {
 		contentToRestore := *content
 		segmentTOC = make(map[int]*toc.SegmentTOC)
@@ -173,8 +175,6 @@ func doRestoreAgent() error {
 		}
 	}
 
-	preloadCreatedPipes(oidList, *copyQueue)
-
 	var currentPipe string
 	for i, oid := range oidList {
 		if wasTerminated {

plugins/example_plugin.bash

@@ -76,6 +76,10 @@ restore_data() {
   echo "restore_data $1 $2" >> /tmp/plugin_out.txt
   filename=`basename "$2"`
   timestamp_dir=`basename $(dirname "$2")`
+  if [ -e "/tmp/GPBACKUP_PLUGIN_FAIL" ] ; then
+	sleep 3
+	echo 'error' >&2
+  fi
   timestamp_day_dir=${timestamp_dir%??????}
 	cat /tmp/plugin_dest/$timestamp_day_dir/$timestamp_dir/$filename
 }