Track cluster-wide Citus version - method 4 - (shmem cache + daemon refresh)

src/backend/distributed/metadata/metadata_cache.c

@@ -63,6 +63,7 @@
 #include "distributed/backend_data.h"
 #include "distributed/citus_depended_object.h"
 #include "distributed/citus_ruleutils.h"
+#include "distributed/cluster_version.h"
 #include "distributed/colocation_utils.h"
 #include "distributed/connection_management.h"
 #include "distributed/foreign_key_relationship.h"
@@ -5135,6 +5136,9 @@ InvalidateNodeRelationCacheCallback(Datum argument, Oid relationId)
 	{
 		workerNodeHashValid = false;
 		LocalNodeId = -1;
+
+		/* the set of nodes changed, so the cached cluster minimum version is stale */
+		InvalidateClusterVersionCache();
 	}
 }
 

src/backend/distributed/operations/cluster_version.c

@@ -0,0 +1,343 @@
+/*-------------------------------------------------------------------------
+ *
+ * cluster_version.c
+ *
+ * UDFs to reason about the Citus version running across the whole cluster,
+ * backed by a shared-memory cache of the computed minimum.
+ *
+ * The minimum cluster version is computed by asking every active primary node
+ * for its own loaded Citus version (citus_version_num()) and taking the
+ * smallest value. Because that fan-out is relatively expensive, the result is
+ * cached in node-local shared memory. The cache is invalidated whenever
+ * pg_dist_node changes (see InvalidateClusterVersionCache), so node additions,
+ * removals and upgrades force a recompute on the next read.
+ *
+ * Note: PostgreSQL shared memory is node-local, so the cache only lives on the
+ * node that computed it; every node maintains its own cache independently.
+ *
+ * Copyright (c) Citus Data, Inc.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+
+#include "storage/ipc.h"
+#include "storage/lwlock.h"
+#include "storage/shmem.h"
+#include "utils/builtins.h"
+
+#include "citus_version.h"
+
+#include "distributed/cluster_version.h"
+#include "distributed/connection_management.h"
+#include "distributed/listutils.h"
+#include "distributed/lock_graph.h"
+#include "distributed/metadata_cache.h"
+#include "distributed/remote_commands.h"
+#include "distributed/worker_manager.h"
+
+#define CLUSTER_VERSION_QUERY "SELECT citus_version_num()"
+
+
+/*
+ * GUC controlling how often the maintenance daemon recomputes and caches the
+ * cluster minimum version via fan-out, in milliseconds. Because the daemon
+ * refreshes the cache proactively, on-demand reads are always served from the
+ * cache and never trigger a fan-out themselves. -1 disables the periodic
+ * refresh.
+ */
+int ClusterVersionRefreshInterval = 60 * 1000;
+
+
+/*
+ * Shared-memory cache of the cluster-wide minimum Citus version. cacheValid is
+ * cleared without holding the lock from the pg_dist_node invalidation callback,
+ * mirroring InvalidateNodeRelationCacheCallback; a stale read is harmless
+ * because the next reader simply recomputes.
+ */
+typedef struct ClusterVersionShmemData
+{
+	NamedLWLockTranche namedLockTranche;
+	LWLock lock;
+	int32 cachedMinVersionNum;
+	bool cacheValid;
+} ClusterVersionShmemData;
+
+
+static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+static ClusterVersionShmemData *ClusterVersionShmem = NULL;
+
+
+PG_FUNCTION_INFO_V1(citus_minimum_cluster_version);
+
+static int32 ComputeMinimumClusterCitusVersion(void);
+static bool TryGetRemoteNodeCitusVersion(WorkerNode *workerNode, int32 *versionNum);
+static text * CitusVersionNumToText(int32 versionNum);
+
+
+/*
+ * citus_minimum_cluster_version returns the oldest (smallest) Citus version
+ * that is running on any active primary node in the cluster, formatted as a
+ * human-readable "major.minor.patch" string (e.g. "15.0.0"). The value is
+ * served from the shared-memory cache when valid, and recomputed otherwise.
+ * Only the integer encoding is cached and compared; the result is decoded to
+ * text for presentation.
+ */
+Datum
+citus_minimum_cluster_version(PG_FUNCTION_ARGS)
+{
+	CheckCitusVersion(ERROR);
+
+	/* fast path: return the cached value if it is still valid */
+	LWLockAcquire(&ClusterVersionShmem->lock, LW_SHARED);
+	bool cacheValid = ClusterVersionShmem->cacheValid;
+	int32 cachedVersion = ClusterVersionShmem->cachedMinVersionNum;
+	LWLockRelease(&ClusterVersionShmem->lock);
+
+	if (cacheValid)
+	{
+		PG_RETURN_TEXT_P(CitusVersionNumToText(cachedVersion));
+	}
+
+	/* slow path: recompute via fan-out and refresh the cache */
+	int32 minimumVersion = ComputeMinimumClusterCitusVersion();
+
+	LWLockAcquire(&ClusterVersionShmem->lock, LW_EXCLUSIVE);
+	ClusterVersionShmem->cachedMinVersionNum = minimumVersion;
+	ClusterVersionShmem->cacheValid = true;
+	LWLockRelease(&ClusterVersionShmem->lock);
+
+	PG_RETURN_TEXT_P(CitusVersionNumToText(minimumVersion));
+}
+
+
+/*
+ * CitusVersionNumToText decodes a citus_version_num() style integer
+ * (major * 10000 + minor * 100 + patch) back into a "major.minor.patch" string,
+ * matching how Citus versions are written (e.g. 120105 -> "12.1.5").
+ */
+static text *
+CitusVersionNumToText(int32 versionNum)
+{
+	int32 major = versionNum / 10000;
+	int32 minor = (versionNum / 100) % 100;
+	int32 patch = versionNum % 100;
+
+	return cstring_to_text(psprintf("%d.%d.%d", major, minor, patch));
+}
+
+
+/*
+ * ComputeMinimumClusterCitusVersion walks over all active primary nodes and
+ * returns the smallest citus_version_num() among them. The local node is
+ * answered from the CITUS_VERSION_NUM constant, every other node is queried
+ * over a connection. If a remote node cannot be reached we error out, because a
+ * minimum that silently ignores unreachable nodes could be used to make an
+ * unsafe decision.
+ */
+static int32
+ComputeMinimumClusterCitusVersion(void)
+{
+	/* the local node always contributes its own loaded version */
+	int32 minimumVersion = CITUS_VERSION_NUM;
+
+	int32 localGroupId = GetLocalGroupId();
+	List *nodeList = ActivePrimaryNodeList(NoLock);
+
+	WorkerNode *workerNode = NULL;
+	foreach_declared_ptr(workerNode, nodeList)
+	{
+		int32 nodeVersion = CITUS_VERSION_NUM;
+
+		if (workerNode->groupId != localGroupId &&
+			!TryGetRemoteNodeCitusVersion(workerNode, &nodeVersion))
+		{
+			ereport(ERROR, (errmsg("could not get Citus version from node \"%s:%d\"",
+								   workerNode->workerName, workerNode->workerPort),
+							errhint("Ensure the node is reachable and running Citus.")));
+		}
+
+		if (nodeVersion < minimumVersion)
+		{
+			minimumVersion = nodeVersion;
+		}
+	}
+
+	return minimumVersion;
+}
+
+
+/*
+ * TryGetRemoteNodeCitusVersion runs citus_version_num() on the given node and,
+ * on success, returns true with *versionNum set. It returns false if the node
+ * could not be queried (e.g. it is unreachable). It never errors on a remote
+ * failure, so it is safe to call from the maintenance daemon.
+ */
+static bool
+TryGetRemoteNodeCitusVersion(WorkerNode *workerNode, int32 *versionNum)
+{
+	int connectionFlags = 0;
+	MultiConnection *connection = GetNodeConnection(connectionFlags,
+												   workerNode->workerName,
+												   workerNode->workerPort);
+
+	PGresult *result = NULL;
+	int executionResult = ExecuteOptionalRemoteCommand(connection,
+													   CLUSTER_VERSION_QUERY, &result);
+
+	if (executionResult != RESPONSE_OKAY || result == NULL || PQntuples(result) != 1)
+	{
+		PQclear(result);
+		ForgetResults(connection);
+		return false;
+	}
+
+	*versionNum = (int32) ParseIntField(result, 0, 0);
+
+	PQclear(result);
+	ForgetResults(connection);
+
+	return true;
+}
+
+
+/*
+ * ClusterVersionShmemSize returns the amount of shared memory needed for the
+ * cluster version cache.
+ */
+Size
+ClusterVersionShmemSize(void)
+{
+	return sizeof(ClusterVersionShmemData);
+}
+
+
+/*
+ * InitializeClusterVersionShmem chains the shared memory startup hook used to
+ * allocate the cluster version cache. Called from _PG_init.
+ */
+void
+InitializeClusterVersionShmem(void)
+{
+	prev_shmem_startup_hook = shmem_startup_hook;
+	shmem_startup_hook = ClusterVersionShmemInit;
+}
+
+
+/*
+ * ClusterVersionShmemInit allocates and initializes the shared memory used for
+ * the cluster version cache.
+ */
+void
+ClusterVersionShmemInit(void)
+{
+	bool alreadyInitialized = false;
+
+	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
+
+	ClusterVersionShmem = (ClusterVersionShmemData *)
+						  ShmemInitStruct("Citus Cluster Version Shmem",
+										  ClusterVersionShmemSize(),
+										  &alreadyInitialized);
+
+	if (!alreadyInitialized)
+	{
+		ClusterVersionShmem->namedLockTranche.trancheName =
+			"Citus Cluster Version Tranche";
+		ClusterVersionShmem->namedLockTranche.trancheId = LWLockNewTrancheId();
+		LWLockRegisterTranche(ClusterVersionShmem->namedLockTranche.trancheId,
+							  ClusterVersionShmem->namedLockTranche.trancheName);
+		LWLockInitialize(&ClusterVersionShmem->lock,
+						 ClusterVersionShmem->namedLockTranche.trancheId);
+
+		ClusterVersionShmem->cachedMinVersionNum = 0;
+		ClusterVersionShmem->cacheValid = false;
+	}
+
+	LWLockRelease(AddinShmemInitLock);
+
+	if (prev_shmem_startup_hook != NULL)
+	{
+		prev_shmem_startup_hook();
+	}
+}
+
+
+/*
+ * InvalidateClusterVersionCache marks the cached minimum version as stale so
+ * that the next reader recomputes it. It is called from the pg_dist_node
+ * relcache invalidation callback. We intentionally do not take the LWLock here,
+ * matching InvalidateNodeRelationCacheCallback, to keep the callback safe and
+ * cheap; a racy stale read only costs one extra recompute.
+ */
+void
+InvalidateClusterVersionCache(void)
+{
+	if (ClusterVersionShmem != NULL)
+	{
+		ClusterVersionShmem->cacheValid = false;
+	}
+}
+
+
+/*
+ * RefreshClusterVersionCache recomputes the cluster minimum version via fan-out
+ * and stores it in the shared-memory cache. It is called periodically by the
+ * maintenance daemon so that on-demand reads are always served from a warm
+ * cache and the value also reflects in-place version changes (which do not
+ * touch pg_dist_node and therefore never trigger InvalidateClusterVersionCache).
+ *
+ * If any node is unreachable we do not store a partial (and therefore possibly
+ * too-high) minimum; instead we invalidate the cache so the next read surfaces
+ * the "node unreachable" error rather than returning a stale value, keeping this
+ * consistent with the strict on-demand path. This never errors on a node being
+ * down, so it keeps the daemon alive. Must run inside a transaction, since the
+ * fan-out reads pg_dist_node and opens connections to the other nodes.
+ */
+void
+RefreshClusterVersionCache(void)
+{
+	if (ClusterVersionShmem == NULL)
+	{
+		return;
+	}
+
+	int32 minimumVersion = CITUS_VERSION_NUM;
+	int32 localGroupId = GetLocalGroupId();
+	List *nodeList = ActivePrimaryNodeList(NoLock);
+
+	WorkerNode *workerNode = NULL;
+	foreach_declared_ptr(workerNode, nodeList)
+	{
+		int32 nodeVersion = CITUS_VERSION_NUM;
+
+		if (workerNode->groupId != localGroupId &&
+			!TryGetRemoteNodeCitusVersion(workerNode, &nodeVersion))
+		{
+			/*
+			 * A node is unreachable so we cannot compute a complete minimum.
+			 * Invalidate the cache rather than storing a partial minimum or
+			 * serving a stale value, so the next read recomputes and surfaces the
+			 * error, matching the strict on-demand path.
+			 */
+			ereport(DEBUG1, (errmsg("invalidating cluster version cache: node "
+									"\"%s:%d\" is unreachable",
+									workerNode->workerName, workerNode->workerPort)));
+			InvalidateClusterVersionCache();
+			return;
+		}
+
+		if (nodeVersion < minimumVersion)
+		{
+			minimumVersion = nodeVersion;
+		}
+	}
+
+	LWLockAcquire(&ClusterVersionShmem->lock, LW_EXCLUSIVE);
+	ClusterVersionShmem->cachedMinVersionNum = minimumVersion;
+	ClusterVersionShmem->cacheValid = true;
+	LWLockRelease(&ClusterVersionShmem->lock);
+}