fix(cluster): spec-5.16 Hardening v1.4 — 2 P1 join-remaster correctness (RF1 P0/8.A + RF2)

Post-ship adversarial review (r5) of the online-join GRD/PCM remaster found two
real defects, both confirmed against the code and fixed in-spec (rule 8.A:
visibility/lock correctness is fail-closed, no forward-link).

RF1 (P0/8.A) -- the JOIN fence bitmap accumulated across episodes and was never
cleared, so a prior rejoiner (now a steady survivor) was wrongly excluded from
the NEXT episode's re-declare barrier -> premature fence lift -> the joiner
cold-serves a block the survivor still holds X on -> double-grant/false-visible.
This overturns the v1.3 CF5 P2 "stale bit is benign, no clear needed" conclusion,
which only reasoned about the over-fence direction and missed the opposite
polarity of the same bitmap in the two barrier-exclusion sites
(cluster_grd_block_view_rebuilt and the WAIT_CLUSTER re-declare ACK barrier).
  Fix: replace join_pcm_fenced_member[words] with a per-node armed-epoch array
  join_pcm_fence_member_epoch[CLUSTER_MAX_NODES].  A node is a recipient of the
  current episode iff member_epoch[node] == fence_epoch; a stale stamp from a
  completed prior episode is < the current epoch and is no longer excluded.
  Per-node monotonic-max keeps the v1.3 concurrent-arm union (no under-fence)
  with no reset race.

RF2 -- qvotec peer-observe counted ANY committed-basis join marker toward the
admission majority and merged max incarnation/epoch across distinct attempts,
unlike the self-admit and startup-seed paths which require a same-commit
majority (INV-J13).  Under a half-publish/retry window distinct-attempt minority
markers could aggregate into a false majority -> a survivor readmits a peer
whose join never durably committed -> membership identity split (8.A).
  Fix: extract the shared selector cluster_join_marker_select_majority() and use
  it in all three sites (peer-observe + self-admit + startup-seed), eliminating
  the third-site drift that caused this.

Tests: cluster_unit test_jr_u17_stale_recipient_not_excluded_next_episode (RF1)
and test_marker_select_majority_groups_by_commit (RF2), both RED->GREEN.

Local gate green: cluster_unit (129 binaries) + cluster_regress 12/12 +
cluster_tap t/325+t/326 (3-node no-double-grant) 35/35 + PG219 219/219 +
clang-format v18 0-diff.  No wire / no catversion change; shmem GRD region only.

Spec: spec-5.16-online-join-grd-pcm-remaster.md (Hardening v1.4)

Author: SqlRush

src/backend/cluster/cluster_grd.c

@@ -632,8 +632,8 @@ cluster_grd_shmem_init(void)
 
 		/* spec-5.16 D2/D3b/D5 — online-join remaster fence + counters. */
 		pg_atomic_init_u64(&cluster_grd_state->join_pcm_fence_epoch, 0);
-		for (i = 0; i < (CLUSTER_MAX_NODES + 63) / 64; i++)
-			pg_atomic_init_u64(&cluster_grd_state->join_pcm_fenced_member[i], 0);
+		for (i = 0; i < CLUSTER_MAX_NODES; i++)
+			pg_atomic_init_u64(&cluster_grd_state->join_pcm_fence_member_epoch[i], 0);
 		pg_atomic_init_u32(&cluster_grd_state->recovery_direction, (uint32)GRD_REMASTER_DIR_NONE);
 		pg_atomic_init_u64(&cluster_grd_state->join_remaster_started_count, 0);
 		pg_atomic_init_u64(&cluster_grd_state->join_remaster_done_count, 0);
@@ -1221,46 +1221,55 @@ cluster_grd_master_map_recompute_for_membership(const uint8 *active_member, uint
  *	Arm the joiner-home PCM block fence SYNCHRONOUSLY (NEVER from the LMON
  *	tick — the async tick is structurally later than the 5.15 write-gate open,
  *	so a fence armed there would leave a "MEMBER-writable but unfenced" window).
- *	Sets join_pcm_fenced_member (the rejoining set) THEN raises join_pcm_fence_
- *	epoch (a write barrier between, so a reader that sees the raised epoch sees
- *	the member set).  Monotonic-max:  a later join re-arms higher;  re-arm of
- *	the same epoch is an idempotent no-op (INV-R12).
+ *	Stamps each rejoining recipient's join_pcm_fence_member_epoch with THIS
+ *	episode's epoch THEN raises join_pcm_fence_epoch (a write barrier between, so
+ *	a reader that sees the raised epoch sees the recipient stamps).  Monotonic-
+ *	max:  a later join re-arms higher;  re-arm of the same epoch is an idempotent
+ *	no-op (INV-R12).
  *
- *	spec-5.16 Hardening v1.3 (Rule 8.A) — the member set is OR-accumulated, never
- *	overwritten.  Two arms race on the rejoining node: qvotec (note_self_admitted,
- *	{self}) and the LMON tick ({evt.join_bitmap} for a multi-joiner episode).  A
- *	plain write would let the {self} arm zero a co-joiner's bit, UNDER-fencing its
- *	home block -> cold-serve -> 8.A double-grant.  OR makes concurrent arms
- *	accumulate the union regardless of interleaving (no under-fence).  A stale bit
- *	carried from a completed prior episode only OVER-fences (the block waits for
- *	this epoch's view_rebuilt barrier, then lifts) — benign liveness, never a
- *	correctness fault, so no per-bit clear is required here.
+ *	spec-5.16 Hardening v1.4 (Rule 8.A) — the recipient set is keyed PER NODE by
+ *	the arming epoch, not an OR-accumulated bitmap.  Two arms race on the
+ *	rejoining node: qvotec (note_self_admitted, {self}) and the LMON tick
+ *	({evt.join_bitmap} for a multi-joiner episode).  Both stamp the SAME epoch on
+ *	their respective nodes, so they union with no lost update and no under-fence.
+ *	Crucially, a node armed in a COMPLETED prior episode keeps its lower stamp,
+ *	which is < the current join_pcm_fence_epoch, so the recipient test (used by
+ *	active_for_shard AND the re-declare barriers) excludes it from THIS episode
+ *	automatically.  The previous v1.3 bitmap was never cleared, so a prior
+ *	rejoiner — now a steady survivor that may hold X on the new joiner's home
+ *	block — was wrongly skipped by the barrier -> premature fence lift -> cold-
+ *	serve -> 8.A double-grant / false-visible (reviewer P1 #1).  Per-node epoch
+ *	keying fixes both the union (no under-fence) and the staleness (no cross-
+ *	episode under-wait) with no reset race.
  */
 void
 cluster_grd_arm_join_pcm_fence(const uint8 *rejoining_set)
 {
 	uint64 epoch;
 	uint64 prev;
-	int w;
+	int node;
 
 	if (cluster_grd_state == NULL || rejoining_set == NULL)
 		return;
 
 	epoch = cluster_epoch_get_current();
 
-	for (w = 0; w < (CLUSTER_MAX_NODES + 63) / 64; w++) {
-		uint64 word = 0;
-		int j;
-
-		for (j = 0; j < 8; j++) {
-			int byte_idx = w * 8 + j;
+	for (node = 0; node < CLUSTER_MAX_NODES; node++) {
+		uint64 cur;
 
-			if (byte_idx < CLUSTER_RECONFIG_DEAD_BITMAP_BYTES)
-				word |= ((uint64)rejoining_set[byte_idx]) << (8 * j);
+		if (node >= CLUSTER_RECONFIG_DEAD_BITMAP_BYTES * 8)
+			break; /* rejoining_set bitmap is exhausted */
+		if (((rejoining_set[node >> 3] >> (node & 7)) & 1) == 0)
+			continue;
+		/* monotonic-max per node — concurrent same-epoch arms union safely */
+		cur = pg_atomic_read_u64(&cluster_grd_state->join_pcm_fence_member_epoch[node]);
+		while (epoch > cur) {
+			if (pg_atomic_compare_exchange_u64(
+					&cluster_grd_state->join_pcm_fence_member_epoch[node], &cur, epoch))
+				break;
 		}
-		pg_atomic_fetch_or_u64(&cluster_grd_state->join_pcm_fenced_member[w], word);
 	}
-	pg_write_barrier(); /* member set visible before the epoch is raised */
+	pg_write_barrier(); /* recipient stamps visible before the epoch is raised */
 
 	prev = pg_atomic_read_u64(&cluster_grd_state->join_pcm_fence_epoch);
 	while (epoch > prev) {
@@ -1286,18 +1295,19 @@ cluster_grd_join_remaster_in_progress(void)
 /*
  * cluster_grd_join_remaster_active_for_shard -- spec-5.16 D3 (INV-R8).
  *
- *	True iff the block's STATIC PCM home (cluster_gcs_lookup_master_static) is
- *	in the armed join_pcm_fenced_member set.  Bound to online_join (the fence
- *	epoch is armed by note_self_admitted / LMON P0-accept), INDEPENDENT of any
- *	GRD master[] movement — so join_remaster_enabled=off still fences (r2 P1-①,
- *	P1-A closure).  false when the fence is not armed.
+ *	True iff the block's STATIC PCM home (cluster_gcs_lookup_master_static) is a
+ *	rejoining RECIPIENT of the CURRENT fence episode (member_epoch[home] ==
+ *	join_pcm_fence_epoch).  Bound to online_join (the fence epoch is armed by
+ *	note_self_admitted / LMON P0-accept), INDEPENDENT of any GRD master[]
+ *	movement — so join_remaster_enabled=off still fences (r2 P1-①, P1-A
+ *	closure).  false when the fence is not armed or the home is a steady member
+ *	(incl. a prior rejoiner whose stamp is from a completed earlier episode).
  */
 bool
 cluster_grd_join_remaster_active_for_shard(BufferTag tag)
 {
 	uint64 fence_epoch;
 	int home;
-	int w, b;
 
 	if (cluster_grd_state == NULL)
 		return false;
@@ -1309,11 +1319,7 @@ cluster_grd_join_remaster_active_for_shard(BufferTag tag)
 	home = cluster_gcs_lookup_master_static(tag);
 	if (home < 0 || home >= CLUSTER_MAX_NODES)
 		return false;
-	w = home >> 6;
-	b = home & 63;
-	return (pg_atomic_read_u64(&cluster_grd_state->join_pcm_fenced_member[w])
-			& (UINT64CONST(1) << b))
-		   != 0;
+	return pg_atomic_read_u64(&cluster_grd_state->join_pcm_fence_member_epoch[home]) == fence_epoch;
 }
 
 /*
@@ -1340,19 +1346,21 @@ cluster_grd_join_remaster_active_for_shard(BufferTag tag)
  *	gate on the joiner is the authoritative backstop (INV-R8/R14).
  */
 
-/* Test whether node_id is in the armed join_pcm_fenced_member (rejoining) set. */
+/*
+ * Test whether node_id is a rejoining RECIPIENT of the fence episode identified
+ * by ref_epoch (member_epoch[node] == ref_epoch).  A stale stamp from a prior
+ * episode (< ref_epoch) returns false, so a now-steady survivor is correctly
+ * waited for by the re-declare barriers (Hardening v1.4, reviewer P1 #1).
+ */
 static inline bool
-join_fenced_member_test(int32 node_id)
+join_fence_is_recipient_for(int32 node_id, uint64 ref_epoch)
 {
-	int w, b;
-
 	if (cluster_grd_state == NULL || node_id < 0 || node_id >= CLUSTER_MAX_NODES)
 		return false;
-	w = node_id >> 6;
-	b = node_id & 63;
-	return (pg_atomic_read_u64(&cluster_grd_state->join_pcm_fenced_member[w])
-			& (UINT64CONST(1) << b))
-		   != 0;
+	if (ref_epoch == 0)
+		return false;
+	return pg_atomic_read_u64(&cluster_grd_state->join_pcm_fence_member_epoch[node_id])
+		   == ref_epoch;
 }
 
 bool
@@ -1379,10 +1387,12 @@ cluster_grd_block_view_rebuilt(BufferTag tag)
 		 * JOIN_COMMITTED event as a reconfig episode (it is published coordinator-
 		 * side only), so it never announces REDECLARE_DONE.  The binding safety
 		 * condition is "every SURVIVOR finished re-declaring its held joiner-home
-		 * blocks" — exclude the fenced set so view_rebuilt converges on the
-		 * survivors' done (Hardening v1.1 + D8 fix).
+		 * blocks" — exclude only THIS episode's recipients so view_rebuilt
+		 * converges on the survivors' done (Hardening v1.1 + D8 fix).  Keyed on
+		 * fence_epoch so a prior rejoiner (now a steady survivor) is NOT skipped
+		 * (Hardening v1.4, reviewer P1 #1: cross-episode under-wait -> 8.A).
 		 */
-		if (join_fenced_member_test(i))
+		if (join_fence_is_recipient_for(i, fence_epoch))
 			continue;
 		if (pg_atomic_read_u64(&cluster_grd_state->recovery_done_epoch[i]) < fence_epoch)
 			return false;
@@ -2074,7 +2084,7 @@ cluster_grd_recovery_lmon_tick(void)
 		 * (P5-P7 below);  for JOIN that rebuilds the joiner's PCM block view even
 		 * when join_remaster_enabled is off.  The two scopes are independent:
 		 * grd_moved_shards = `affected` here (GRD, GUC-gated); pcm_fenced_home_set
-		 * lives in join_pcm_fenced_member (PCM, online_join-gated, armed at P0).
+		 * lives in join_pcm_fence_member_epoch (PCM, online_join-gated, armed at P0).
 		 */
 		memset(affected, 0, sizeof(affected));
 		if (pg_atomic_read_u32(&cluster_grd_state->recovery_direction)
@@ -2307,10 +2317,14 @@ cluster_grd_recovery_lmon_tick(void)
 				 * and never observes the JOIN_COMMITTED event as its own reconfig
 				 * episode (it is published coordinator-side only), so it never
 				 * announces REDECLARE_DONE.  Waiting for it would wedge the survivor's
-				 * barrier forever.  The survivors (everyone outside the fenced set)
-				 * ARE the re-declarers and must converge.
+				 * barrier forever.  The survivors (everyone outside THIS episode's
+				 * recipient set) ARE the re-declarers and must converge.  Keyed on
+				 * episode_epoch so a prior rejoiner (now a steady survivor) is still
+				 * waited for (Hardening v1.4, reviewer P1 #1: a stale cross-episode
+				 * exclusion would skip a survivor holding X on the joiner's home
+				 * block -> premature unfreeze -> 8.A double-grant).
 				 */
-				if (is_join && join_fenced_member_test(i))
+				if (is_join && join_fence_is_recipient_for(i, episode_epoch))
 					continue;
 				if (pg_atomic_read_u64(&cluster_grd_state->recovery_done_epoch[i])
 					< episode_epoch) {

src/backend/cluster/cluster_membership.c

@@ -291,6 +291,41 @@ cluster_join_marker_same_commit(const ClusterJoinCommitMarker *a, const ClusterJ
 	return memcmp(a, b, offsetof(ClusterJoinCommitMarker, crc32c)) == 0;
 }
 
+/*
+ * INV-J13 majority selector (shared by self-admit, startup-seed and qvotec
+ * peer-observe).  `markers` must already be committed-basis; O(n^2) over n disks
+ * (n <= CLUSTER_MAX_VOTING_DISKS, small).  Returns the index of the first marker
+ * that is same_commit with >= `majority` of the array (i.e. a single commit
+ * attempt that actually reached a disk majority), or -1.  A set of distinct-
+ * attempt minority markers therefore selects NOTHING — they never aggregate
+ * into a false majority (reviewer P1 #2 / P1-3).
+ */
+int
+cluster_join_marker_select_majority(const ClusterJoinCommitMarker *markers, int n, uint32 majority,
+									uint32 *out_agree)
+{
+	int a, b;
+
+	if (out_agree != NULL)
+		*out_agree = 0;
+	if (markers == NULL || n <= 0)
+		return -1;
+
+	for (a = 0; a < n; a++) {
+		uint32 same = 0;
+
+		for (b = 0; b < n; b++)
+			if (cluster_join_marker_same_commit(&markers[a], &markers[b]))
+				same++;
+		if (same >= majority) {
+			if (out_agree != NULL)
+				*out_agree = same;
+			return a;
+		}
+	}
+	return -1;
+}
+
 /*
  * Apply one durable marker to the admitted floor (INV-J7).  Only a committed
  * basis raises the floor; record_admitted is monotonic so re-applying / lower

src/backend/cluster/cluster_qvotec.c

@@ -879,17 +879,27 @@ qvotec_poll_once(void)
 	 * joins the re-declare barrier (without it the coordinator's barrier waits
 	 * forever for a non-participating survivor in >=3-node).  Mirrors the self-
 	 * admission read below; one extra region-3 slot read per (disk × peer) per poll
-	 * (qvotec already has the fds).  Majority + COMMITTED-basis validated; a never-
-	 * joined / undeclared slot fails the basis check and publishes nothing.
+	 * (qvotec already has the fds).  Same-commit majority + COMMITTED-basis
+	 * validated (INV-J13); a never-joined / undeclared slot fails the basis check
+	 * and publishes nothing.
+	 *
+	 * Hardening v1.4 (reviewer P1 #2) — this used to count ANY committed-basis
+	 * marker toward the majority and take the max incarnation/epoch, UNLIKE the
+	 * self-admit and startup-seed paths.  Two distinct join attempts (different
+	 * coordinator / epoch / nonce), each on a minority of disks, would then
+	 * aggregate into a false majority -> the survivor readmits a peer whose join
+	 * never durably committed (DEAD->MEMBER + observer JOIN event) -> 8.A.  Now it
+	 * mirrors them exactly: collect committed-basis markers, then require a
+	 * same-commit majority via the shared cluster_join_marker_select_majority.
 	 */
 	{
 		uint32 node;
 		uint32 majority = ((uint32)qvotec_n_disks / 2u) + 1u;
 
 		for (node = 0; node < CLUSTER_MAX_NODES; node++) {
-			uint32 agree = 0;
-			uint64 best_incarnation = 0;
-			uint64 best_admitted_epoch = 0;
+			ClusterJoinCommitMarker committed[CLUSTER_MAX_VOTING_DISKS];
+			int n_committed = 0;
+			int win;
 			int d;
 
 			if ((int32)node == cluster_node_id)
@@ -908,15 +918,13 @@ qvotec_poll_once(void)
 				memcpy(&m, jslot.bytes, sizeof(m));
 				if (!cluster_join_marker_is_committed_basis(&m, (int32)node))
 					continue;
-				agree++;
-				if (m.admitted_incarnation > best_incarnation)
-					best_incarnation = m.admitted_incarnation;
-				if (m.admitted_epoch > best_admitted_epoch)
-					best_admitted_epoch = m.admitted_epoch;
+				committed[n_committed++] = m;
 			}
-			if (agree >= majority && best_incarnation > 0)
-				cluster_reconfig_record_observed_committed_join((int32)node, best_incarnation,
-																best_admitted_epoch);
+			win = cluster_join_marker_select_majority(committed, n_committed, majority, NULL);
+			if (win >= 0 && committed[win].admitted_incarnation > 0)
+				cluster_reconfig_record_observed_committed_join((int32)node,
+																committed[win].admitted_incarnation,
+																committed[win].admitted_epoch);
 		}
 	}
 
@@ -940,9 +948,8 @@ qvotec_poll_once(void)
 		ClusterJoinCommitMarker self_markers[CLUSTER_MAX_VOTING_DISKS];
 		int n_self = 0;
 		uint32 majority = ((uint32)qvotec_n_disks / 2u) + 1u;
-		int d, a, b;
-		int win = -1;
-		uint32 win_agree = 0;
+		int d;
+		int win;
 
 		for (d = 0; d < qvotec_n_disks; d++) {
 			union {
@@ -962,23 +969,12 @@ qvotec_poll_once(void)
 				continue; /* a stale prior-incarnation admission — not us */
 			self_markers[n_self++] = m;
 		}
-		/* HF-3: find a single commit (nonce) present on >= majority disks. */
-		for (a = 0; a < n_self; a++) {
-			uint32 same = 0;
-
-			for (b = 0; b < n_self; b++)
-				if (cluster_join_marker_same_commit(&self_markers[a], &self_markers[b]))
-					same++;
-			if (same >= majority) {
-				win = a;
-				win_agree = same;
-				break;
-			}
-		}
+		/* HF-3 / INV-J13: find a single commit (nonce) present on >= majority disks
+		 * (shared selector — same logic as startup-seed and peer-observe). */
+		win = cluster_join_marker_select_majority(self_markers, n_self, majority, NULL);
 
 		/* HF-1: gate-open requires the publish-proof too, not the marker alone. */
-		if (win >= 0 && win_agree >= majority
-			&& cluster_reconfig_join_publish_proven(self_markers[win].admitted_epoch)) {
+		if (win >= 0 && cluster_reconfig_join_publish_proven(self_markers[win].admitted_epoch)) {
 			cluster_reconfig_note_self_admitted(self_markers[win].admitted_epoch);
 			/*
 			 * spec-5.16 (3-node rejoin) — the same durable COMMITTED join marker

src/backend/cluster/cluster_reconfig.c

@@ -2383,9 +2383,9 @@ cluster_membership_seed_last_admitted_from_voting_disk(const int *fds, int n_dis
 		} slot;
 		ClusterJoinCommitMarker committed[CLUSTER_MAX_VOTING_DISKS];
 		int n_committed = 0;
-		int win = -1;
+		int win;
 		uint32 win_agree = 0;
-		int d, a, b;
+		int d;
 
 		if (cluster_conf_lookup_node(s) == NULL)
 			continue;
@@ -2408,21 +2408,10 @@ cluster_membership_seed_last_admitted_from_voting_disk(const int *fds, int n_dis
 		 * nonce), not "any COMMITTED marker".  Two minority writes from
 		 * different attempts (different coordinator / epoch) must not aggregate.
 		 * Only a marker that actually reached a disk majority represents a real
-		 * admission, so only it raises the monotonic floor (P1-3).  O(disks^2),
-		 * disks <= CLUSTER_MAX_VOTING_DISKS (small).
+		 * admission, so only it raises the monotonic floor (P1-3).  Shared
+		 * selector — same logic as self-admit and qvotec peer-observe.
 		 */
-		for (a = 0; a < n_committed; a++) {
-			uint32 same = 0;
-
-			for (b = 0; b < n_committed; b++)
-				if (cluster_join_marker_same_commit(&committed[a], &committed[b]))
-					same++;
-			if (same >= majority) {
-				win = a;
-				win_agree = same;
-				break;
-			}
-		}
+		win = cluster_join_marker_select_majority(committed, n_committed, majority, &win_agree);
 
 		if (win >= 0 && committed[win].admitted_incarnation > 0) {
 			uint64 win_incarnation = committed[win].admitted_incarnation;

src/include/cluster/cluster_gcs_block.h

@@ -1010,10 +1010,11 @@ extern ClusterGcsBlockPhase cluster_gcs_block_phase_for_tag(BufferTag tag);
  * requester-side phase gate and the master-side envelope handler consume them).
  *
  *	cluster_grd_join_remaster_active_for_shard:  the block's STATIC PCM home
- *	    (cluster_gcs_lookup_master_static) is in the armed join_pcm_fenced_member
- *	    set (bound to online_join, INDEPENDENT of any GRD master[] movement —
- *	    so join_remaster_enabled=off still fences, r2 P1-①).  false when the
- *	    fence is not armed (join_pcm_fence_epoch == 0).
+ *	    (cluster_gcs_lookup_master_static) is a rejoining RECIPIENT of the current
+ *	    fence episode (join_pcm_fence_member_epoch[home] == join_pcm_fence_epoch;
+ *	    bound to online_join, INDEPENDENT of any GRD master[] movement — so
+ *	    join_remaster_enabled=off still fences, r2 P1-①).  false when the fence is
+ *	    not armed (join_pcm_fence_epoch == 0) or the home is a steady member.
  *	cluster_grd_block_view_rebuilt:  the joiner-home view is rebuilt — i.e.
  *	    EVERY declared member's recovery_done_epoch >= join_pcm_fence_epoch
  *	    (Hardening v1.1:  the all-members all_done barrier, NOT the joiner's own