fix: a cancelled remote parallel export aborts its workers' uploads (#1722)

.github/workflows/ci.yml

@@ -185,6 +185,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and refresh the package lists
         run: |
           set -euo pipefail

.github/workflows/nightly.yml

@@ -745,6 +745,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and install PostgreSQL ${{ matrix.pg }} with its headers
         run: |
           set -euo pipefail

objstore/columnar_objstore_module.c

@@ -245,8 +245,15 @@ os_wait(PgColumnarObjHandle *h, uint32 sockEvents, int *quiet)
 {
 	int			rc;
 
+	/*
+	 * WL_TIMEOUT, or the slice is not a slice: without it WaitLatchOrSocket
+	 * ignores the timeout and waits for the socket or the latch alone, so a
+	 * silent peer never counts a quiet slice. Under proc_exit (a terminated
+	 * worker aborting its upload, #1727) interrupts are held and the latch
+	 * alone never ends the wait either.
+	 */
 	rc = WaitLatchOrSocket(MyLatch,
-						   WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | sockEvents,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH | sockEvents,
 						   h->fd, OS_WAIT_SLICE_MS, PG_WAIT_EXTENSION);
 	if (rc & WL_LATCH_SET)
 		ResetLatch(MyLatch);
@@ -2002,8 +2009,8 @@ os_flush_part(PgColumnarObjSink *s)
 						s->h->url, resp.status)));
 	/* The part's ETag is the server's word for its bytes; CompleteMultipart
 	 * validates the list against it on a real S3, so echo it verbatim (already
-	 * quoted by the server). The fixture ignores it and concatenates by part
-	 * number, so both are satisfied. */
+	 * quoted by the server). The fixture checks it the same way and assembles
+	 * the listed parts, so both are satisfied. */
 	etag = resp.etag[0] ? resp.etag : "\"\"";
 	appendStringInfo(&s->completeXml,
 					 "<Part><PartNumber>%d</PartNumber><ETag>%s</ETag></Part>",
@@ -2170,6 +2177,14 @@ objstore_sink_abort(PgColumnarObjSink *s)
 			char	   *encId = os_uriencode_query(s->uploadId);
 			char	   *query = psprintf("uploadId=%s", encId);
 
+			/*
+			 * The abort can land mid-request (a cancel or a SIGTERM inside
+			 * os_wait), with a part's body still on this connection: the
+			 * DELETE would be read as more of that body, or its answer
+			 * confused with the part's. A fresh connection makes it a request
+			 * of its own (#1722).
+			 */
+			os_disconnect(s->h);
 			os_write_request(s->h, "DELETE", query, NULL, 0, &resp, NULL);
 			pfree(query);
 			pfree(encId);

src/columnar_sink.c

@@ -16,8 +16,11 @@
 
 #include <unistd.h>
 
+#include "access/xact.h"
+#include "lib/ilist.h"
 #include "miscadmin.h"
 #include "storage/fd.h"
+#include "storage/ipc.h"
 
 #include "columnar_objstore.h"
 #include "columnar_sink.h"
@@ -34,8 +37,62 @@ struct PqSink
 	/* remote (#394): NULL api => local sink */
 	const PgColumnarObjStoreApi *api;
 	PgColumnarObjSink *rsink;
+	dlist_node	openNode;		/* on sink_open_remote while rsink is live */
 };
 
+/*
+ * Remote sinks with an upload in progress (#1722). A FATAL exit (SIGTERM, as
+ * pexport_cleanup sends each parallel_export worker on cancel, or
+ * pg_terminate_backend) calls proc_exit without unwinding, so the writers'
+ * PG_CATCH never reaches PgColumnarSinkAbort and the multipart upload stays
+ * open on the store, billed until a lifecycle rule reaps it. A
+ * before_shmem_exit callback aborts whatever is still on this list. It is
+ * registered after InitPostgres's ShutdownPostgres, so it runs first, while
+ * the transaction and the module's handles are still intact. Transaction end
+ * empties the list: every writer finishes or aborts its sink inside its own
+ * statement, so an entry still listed then belongs to memory being released.
+ */
+static dlist_head sink_open_remote = DLIST_STATIC_INIT(sink_open_remote);
+static bool sink_exit_registered = false;
+
+static void
+sink_abort_open_at_exit(int code, Datum arg)
+{
+	dlist_mutable_iter it;
+
+	/*
+	 * The module's abort never raises through PG_TRY, but under proc_exit an
+	 * ERROR is promoted to FATAL, which no CATCH sees. A failed DELETE then
+	 * ends this callback with one more FATAL line; shmem_exit has already
+	 * popped it, so the remaining callbacks still run.
+	 */
+	dlist_foreach_modify(it, &sink_open_remote)
+	{
+		PqSink	   *snk = dlist_container(PqSink, openNode, it.cur);
+
+		dlist_delete(&snk->openNode);
+		snk->api->sink_abort(snk->rsink);
+		snk->rsink = NULL;
+	}
+}
+
+static void
+sink_xact_callback(XactEvent event, void *arg)
+{
+	switch (event)
+	{
+		case XACT_EVENT_COMMIT:
+		case XACT_EVENT_PARALLEL_COMMIT:
+		case XACT_EVENT_ABORT:
+		case XACT_EVENT_PARALLEL_ABORT:
+		case XACT_EVENT_PREPARE:
+			dlist_init(&sink_open_remote);
+			break;
+		default:
+			break;
+	}
+}
+
 PqSink *
 PgColumnarSinkOpenLocal(const char *path)
 {
@@ -83,6 +140,13 @@ PgColumnarSinkOpen(const char *path)
 		snk->finalPath = pstrdup(path);
 		snk->api = api;
 		snk->rsink = api->sink_create(path, NULL);
+		if (!sink_exit_registered)
+		{
+			before_shmem_exit(sink_abort_open_at_exit, (Datum) 0);
+			RegisterXactCallback(sink_xact_callback, NULL);
+			sink_exit_registered = true;
+		}
+		dlist_push_head(&sink_open_remote, &snk->openNode);
 		return snk;
 	}
 	return PgColumnarSinkOpenLocal(path);
@@ -140,6 +204,8 @@ PgColumnarSinkFinish(PqSink *snk)
 	if (snk->api != NULL)
 	{
 		snk->api->sink_finish(snk->rsink);
+		/* only now: a complete that raised leaves it listed for the abort */
+		dlist_delete(&snk->openNode);
 		snk->rsink = NULL;
 		return;
 	}
@@ -173,6 +239,7 @@ PgColumnarSinkAbort(PqSink *snk)
 	{
 		if (snk->rsink != NULL)
 		{
+			dlist_delete(&snk->openNode);
 			snk->api->sink_abort(snk->rsink);
 			snk->rsink = NULL;
 		}