From 983d24cf68a8afd215a5a14d98fffabd41cc4cff Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Fri, 11 Sep 2026 16:09:06 +0000 Subject: [PATCH] Prevent double frees when retrying interrupted dequeues Clear each slot reference before resolving and releasing its string, so dequeue recovery cannot free error text or release query text twice Attach to DSA before clearing references, since attachment can raise ERROR Delay storing enqueue string references until just before advancing head Consolidate DSA attachment helpers and remove ineffective exception, handling around noexcept exporter teardown --- src/export/stats_exporter.cc | 13 +++---------- src/queue/psch_dsa.c | 37 +++++++++++++----------------------- src/queue/psch_dsa.h | 18 +++++++----------- src/queue/query_intern.c | 6 +++--- src/queue/query_intern.h | 2 +- src/queue/shmem.c | 36 +++++++++++++++++++++++------------ src/worker/bgworker.c | 3 ++- 7 files changed, 53 insertions(+), 62 deletions(-) diff --git a/src/export/stats_exporter.cc b/src/export/stats_exporter.cc index 6f38843..1ac0a12 100644 --- a/src/export/stats_exporter.cc +++ b/src/export/stats_exporter.cc @@ -549,17 +549,10 @@ int PschGetConsecutiveFailures(void) { } } -// Exception barrier: the OTel exporter's destructors (gRPC stub teardown, -// protobuf arena release) can throw. Catching here prevents the throw from -// crossing the on_proc_exit chain. +// Destructors and unique_ptr::reset() are noexcept, a throw during teardown +// reaches PschTerminateHandler through std::terminate, never this frame void PschExporterShutdown(void) { - try { - g_exporter.exporter.reset(); - } catch (const std::bad_alloc&) { - LogExporterWarning("exporter shutdown", "out of memory"); - } catch (const std::exception& e) { - LogExporterWarning("exporter shutdown exception", e.what()); - } + g_exporter.exporter.reset(); elog(LOG, "pg_stat_ch: statistics exporter shutdown"); } diff --git a/src/queue/psch_dsa.c b/src/queue/psch_dsa.c index 612cdb0..b204b66 100644 --- a/src/queue/psch_dsa.c +++ b/src/queue/psch_dsa.c @@ -53,12 +53,12 @@ void PschDsaInit(PschSharedState* state, void* dsa_place) { dsa_detach(dsa); // Postmaster detaches; backends/bgworker re-attach } -void PschDsaAttach(void) { +dsa_area* PschDsaAttach(void) { if (psch_dsa != NULL) { - return; + return psch_dsa; } if (psch_shared_state == NULL || psch_shared_state->raw_dsa_area == NULL) { - return; + return NULL; } // Attach in TopMemoryContext so the dsa_area handle survives transaction // boundaries. Without this, the palloc'd dsa_area struct would be freed @@ -73,12 +73,6 @@ void PschDsaAttach(void) { on_shmem_exit(PschDsaDetachOnExit, 0); psch_dsa_exit_hook_registered = true; } -} - -dsa_area* PschDsaGetArea(void) { - if (psch_dsa == NULL) { - PschDsaAttach(); - } return psch_dsa; } @@ -86,7 +80,7 @@ dsa_pointer PschDsaAllocString(const char* src, uint16 len, uint16 max_len) { if (len == 0) { return InvalidDsaPointer; } - dsa_area* dsa = PschDsaGetArea(); + dsa_area* dsa = PschDsaAttach(); if (dsa == NULL) { return InvalidDsaPointer; } @@ -104,21 +98,16 @@ dsa_pointer PschDsaAllocString(const char* src, uint16 len, uint16 max_len) { void PschDsaResolveString(dsa_pointer dp, uint16 src_len, char* dst_buf, uint16 max_len, uint16* out_len) { - if (DsaPointerIsValid(dp)) { - dsa_area* dsa = PschDsaGetArea(); - if (dsa == NULL) { - dst_buf[0] = '\0'; - *out_len = 0; - return; - } - char* src = (char*)dsa_get_address(dsa, dp); - uint16 len = Min(src_len, (uint16)(max_len - 1)); - memcpy(dst_buf, src, len); - dst_buf[len] = '\0'; - *out_len = len; - dsa_free(dsa, dp); - } else { + dsa_area* dsa = DsaPointerIsValid(dp) ? PschDsaAttach() : NULL; + if (dsa == NULL) { dst_buf[0] = '\0'; *out_len = 0; + return; } + char* src = (char*)dsa_get_address(dsa, dp); + uint16 len = Min(src_len, (uint16)(max_len - 1)); + memcpy(dst_buf, src, len); + dst_buf[len] = '\0'; + *out_len = len; + dsa_free(dsa, dp); } diff --git a/src/queue/psch_dsa.h b/src/queue/psch_dsa.h index 77db84f..5a02c83 100644 --- a/src/queue/psch_dsa.h +++ b/src/queue/psch_dsa.h @@ -29,9 +29,9 @@ // LIFECYCLE // ========= // 1. Postmaster: PschDsaInit() — dsa_create_in_place, dsa_pin, dsa_detach -// 2. Backends: PschDsaAllocString() calls PschDsaAttach() lazily on first use -// 3. Bgworker: PschDsaAttach() eagerly at startup, then PschDsaResolveString() -// on each dequeue (resolves pointer + frees DSA memory) +// 2. Backends: PschDsaAllocString() attaches on first enqueue +// 3. Bgworker: Attach before consuming queue slots +// String helpers reuse process-local attachment until process exit // // CONCURRENCY // =========== @@ -103,14 +103,10 @@ Size PschDsaShmemSize(void); // PschDsaShmemSize() bytes available. void PschDsaInit(PschSharedState* state, void* dsa_place); -// Attach to the DSA area (lazy, idempotent). -// Must be called before PschDsaAllocString / PschDsaResolveString. -// Backends attach lazily on first enqueue; bgworker attaches eagerly at startup. -void PschDsaAttach(void); - -// Get the process-local DSA handle, attaching lazily if needed. -// Returns nullptr if shared state / raw_dsa_area is unavailable. -dsa_area* PschDsaGetArea(void); +// Attach once per process, return existing handle on subsequent calls +// Return NULL if shared state or raw_dsa_area is unavailable +// Call before clearing shared references, attachment can raise ERROR +dsa_area* PschDsaAttach(void); // Allocate a DSA string from an inline buffer. // Returns InvalidDsaPointer on zero-length, unavailable DSA handle, or allocation diff --git a/src/queue/query_intern.c b/src/queue/query_intern.c index 9ea9cf7..5e10e80 100644 --- a/src/queue/query_intern.c +++ b/src/queue/query_intern.c @@ -180,7 +180,7 @@ dsa_pointer PschQueryInternAcquire(Oid dbid, uint64 queryid, const char* query, return InvalidDsaPointer; } - dsa = PschDsaGetArea(); + dsa = PschDsaAttach(); if (dsa == NULL) { return InvalidDsaPointer; } @@ -264,7 +264,7 @@ static void ReleaseRef(dsa_pointer ref) { return; } - dsa = PschDsaGetArea(); + dsa = PschDsaAttach(); if (dsa == NULL) { return; } @@ -315,7 +315,7 @@ static void ResolveInto(dsa_pointer ref, char* dst, uint16 dst_size, uint16* out return; } - dsa = PschDsaGetArea(); + dsa = PschDsaAttach(); if (dsa == NULL) { return; } diff --git a/src/queue/query_intern.h b/src/queue/query_intern.h index 148d773..3f9fd6e 100644 --- a/src/queue/query_intern.h +++ b/src/queue/query_intern.h @@ -64,7 +64,7 @@ void PschQueryInternShmemInit(LWLockPadded* lwlock_base); // (collisions are treated as a miss with no insert — exporting empty query // text is preferable to exporting the wrong SQL). // -// Attaches lazily via PschDsaGetArea; returns InvalidDsaPointer if DSA is +// Attaches lazily via PschDsaAttach; returns InvalidDsaPointer if DSA is // unavailable. dsa_pointer PschQueryInternAcquire(Oid dbid, uint64 queryid, const char* query, uint16 query_len); diff --git a/src/queue/shmem.c b/src/queue/shmem.c index b3bb29f..966888d 100644 --- a/src/queue/shmem.c +++ b/src/queue/shmem.c @@ -114,10 +114,12 @@ static bool TryEnqueueLocked(const PschEvent* event, uint32 capacity) { // client_addr, lengths — everything before the variable-length data). memcpy(slot, event, kFixedPrefixSize); - // 2. Allocate DSA for err_message and query text - slot->err_message_dsa = + // 2. Allocate DSA for err_message and query text. Store pointers in slot + // only right before head advances, so slots outside [tail, head) never + // hold references + dsa_pointer err_message_dsa = PschDsaAllocString(event->err_message, event->err_message_len, PSCH_MAX_ERR_MSG_LEN); - if (event->err_message_len > 0 && !DsaPointerIsValid(slot->err_message_dsa)) { + if (event->err_message_len > 0 && !DsaPointerIsValid(err_message_dsa)) { slot->err_message_len = 0; // Lost string on OOM — numeric data preserved } @@ -125,21 +127,22 @@ static bool TryEnqueueLocked(const PschEvent* event, uint32 capacity) { // normalized queries share a single DSA-allocated body. See query_intern.h // for the design rationale. On miss + DSA OOM, miss + hash-full, or hash // collision we drop the query bytes (numeric data is preserved). + dsa_pointer query_dsa = InvalidDsaPointer; if (event->query_len > 0) { // Clamp the input length to what we'd have stored anyway, so the intern // key doesn't include trailing bytes the consumer would have truncated. uint16 clamped_len = Min(event->query_len, (uint16)(PSCH_MAX_QUERY_LEN - 1)); - slot->query_dsa = - PschQueryInternAcquire(event->dbid, event->queryid, event->query, clamped_len); - if (!DsaPointerIsValid(slot->query_dsa)) { + query_dsa = PschQueryInternAcquire(event->dbid, event->queryid, event->query, clamped_len); + if (!DsaPointerIsValid(query_dsa)) { slot->query_len = 0; } else { slot->query_len = clamped_len; } - } else { - slot->query_dsa = InvalidDsaPointer; } + slot->err_message_dsa = err_message_dsa; + slot->query_dsa = query_dsa; + // CRITICAL: Memory barrier ensures the event data is written to shared memory // before we update head. Without this, the consumer might read stale data on // weakly-ordered architectures (ARM, PowerPC). Pattern from shm_mq.c. @@ -441,15 +444,24 @@ bool PschDequeueEvent(PschEvent* event) { uint32 mask = capacity - 1; PschRingEntry* slot = &GetRingBuffer()[tail & mask]; + // Attach up front, a lazy attach after step 2 detaches the references could raise + PschDsaAttach(); + // 1. Copy the entire fixed-field prefix (all numeric fields, app_name, // client_addr, lengths — everything before the variable-length data). memcpy(event, slot, kFixedPrefixSize); - // 2. Resolve err_message (per-event DSA) and query text (shared interner). - PschDsaResolveString(slot->err_message_dsa, slot->err_message_len, event->err_message, + // 2. Clear each reference before consuming it. ExportBatchWithRecovery + // catches ERROR and retries this slot, so a stale pointer would be freed + // twice. An abort mid-consume leaks at most one reference + dsa_pointer err_message_dsa = slot->err_message_dsa; + slot->err_message_dsa = InvalidDsaPointer; + PschDsaResolveString(err_message_dsa, event->err_message_len, event->err_message, PSCH_MAX_ERR_MSG_LEN, &event->err_message_len); - PschQueryInternResolveAndRelease(slot->query_dsa, event->query, PSCH_MAX_QUERY_LEN, - &event->query_len); + + dsa_pointer query_dsa = slot->query_dsa; + slot->query_dsa = InvalidDsaPointer; + PschQueryInternResolveAndRelease(query_dsa, event->query, PSCH_MAX_QUERY_LEN, &event->query_len); // CRITICAL: Write barrier ensures all reads and DSA frees complete before we // update tail. Producers cannot reuse this slot until tail advances past it. diff --git a/src/worker/bgworker.c b/src/worker/bgworker.c index 5f0a02b..09f2aef 100644 --- a/src/worker/bgworker.c +++ b/src/worker/bgworker.c @@ -202,7 +202,8 @@ void PschBgworkerMain(Datum main_arg pg_attribute_unused()) { // Store our PID for signaling (used by pg_stat_ch_flush()) PschSetBgworkerPid(MyProcPid); - // Attach to DSA area eagerly so the first dequeue doesn't hit lazy init + // Attach in this C frame: an attach ERROR raised inside DequeueEvents would + // longjmp past its std::vector destructor PschDsaAttach(); elog(LOG, "pg_stat_ch: background worker started (pid=%d)", MyProcPid);