Merge branch 'max-netconf-queue' into no-migrate

Helps prevent lost messages.  Commbied with a smaller limit of messages
cached in memory before processing

nonfree/controller/CentralDB.cpp

@@ -43,6 +43,26 @@ using Attrs = std::vector<std::pair<std::string, std::string> >;
 using Item = std::pair<std::string, Attrs>;
 using ItemStream = std::vector<Item>;
 
+namespace ZeroTier {
+
+// Per-thread hand-off backing setPendingCompletion()/takePendingCompletion() (declared in
+// NotificationListener.hpp). The PubSub subscriber callback sets it around its synchronous
+// onNotification() call; the commit-queue enqueue path (save/eraseNetwork/eraseMember) takes
+// it when it defers work, moving ownership into the _queueItem so the commit thread can ack.
+static thread_local std::shared_ptr<NotificationCompletion> t_pendingCompletion;
+
+void setPendingCompletion(std::shared_ptr<NotificationCompletion> completion)
+{
+	t_pendingCompletion = std::move(completion);
+}
+
+std::shared_ptr<NotificationCompletion> takePendingCompletion()
+{
+	return std::move(t_pendingCompletion);	 // a moved-from shared_ptr is left empty
+}
+
+}	// namespace ZeroTier
+
 CentralDB::CentralDB(const Identity& myId,
 					 const char* connString,
 					 int listenPort,
@@ -306,6 +326,9 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 					_queueItem qi;
 					qi.jsonData = record;
 					qi.notifyListeners = notifyListeners;
+					// Take the deferred ack (if any) so the commit thread acks/nacks this change
+					// only after the DB write; null for internal (non-PubSub) callers.
+					qi.completion = takePendingCompletion();
 					OtelCarrier<std::map<std::string, std::string> > carrier(qi.traceContext);
 					auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 					auto propagator =
@@ -331,6 +354,9 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 					_queueItem qi;
 					qi.jsonData = record;
 					qi.notifyListeners = notifyListeners;
+					// Take the deferred ack (if any) so the commit thread acks/nacks this change
+					// only after the DB write; null for internal (non-PubSub) callers.
+					qi.completion = takePendingCompletion();
 					OtelCarrier<std::map<std::string, std::string> > carrier(qi.traceContext);
 					auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 					auto propagator =
@@ -375,6 +401,7 @@ void CentralDB::eraseNetwork(const uint64_t networkId)
 	qi.jsonData["id"] = tmp2;
 	qi.jsonData["objtype"] = "_delete_network";
 	qi.notifyListeners = true;
+	qi.completion = takePendingCompletion();
 	OtelCarrier<std::map<std::string, std::string> > carrier(qi.traceContext);
 	auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 	auto propagator = opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
@@ -403,6 +430,7 @@ void CentralDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
 	qi.jsonData["id"] = memberIdStr;
 	qi.jsonData["objtype"] = "_delete_member";
 	qi.notifyListeners = true;
+	qi.completion = takePendingCompletion();
 	OtelCarrier<std::map<std::string, std::string> > carrier(qi.traceContext);
 	auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 	auto propagator = opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
@@ -912,7 +940,6 @@ void CentralDB::initializeMembers()
 				   >
 			row;
 
-		auto tmp = std::chrono::high_resolution_clock::now();
 		uint64_t count = 0;
 		uint64_t total = 0;
 		while (stream >> row) {
@@ -1135,11 +1162,11 @@ void CentralDB::heartbeat()
 
 void CentralDB::_requeueFailedCommit(_queueItem& qitem)
 {
-	// The PubSub message that produced this change was already acked when it was
-	// enqueued, so a failed DB write would otherwise be lost.  Re-queue it (with
-	// capped backoff) so transient failures -- e.g. an AlloyDB maintenance restart
-	// -- are retried rather than dropped.  A genuinely poison item is dropped after
-	// a bounded number of attempts so it can't spin a commit thread forever.
+	// Reached only for INTERNAL (non-PubSub) writes -- items carrying a delivery completion are
+	// nacked for redelivery by _finishCommit instead. Internal writes have no redelivery, so
+	// re-queue them (with capped backoff) to ride out transient DB failures (e.g. an AlloyDB
+	// maintenance restart) rather than dropping them. A genuinely poison item is dropped after a
+	// bounded number of attempts so it can't spin a commit thread forever.
 	if (_run != 1) {
 		// shutting down -- don't requeue
 		return;
@@ -1167,6 +1194,23 @@ void CentralDB::_requeueFailedCommit(_queueItem& qitem)
 	_commitQueue.post(qitem);
 }
 
+void CentralDB::_finishCommit(_queueItem& qitem, NotificationResult result)
+{
+	if (qitem.completion) {
+		// Delivery-backed (PubSub): ack on success / not-applicable, nack to redeliver on a
+		// transient failure. This replaces re-queuing for these items -- Pub/Sub's own
+		// redelivery handles retries, and a persistently failing message is routed to the
+		// dead-letter topic by the subscription policy rather than dropped in-memory.
+		qitem.completion->complete(result);
+		return;
+	}
+	// Internal write (no delivery message): keep the in-memory retry path for transient
+	// failures; nothing to do on success.
+	if (result == NotificationResult::TransientFailure) {
+		_requeueFailedCommit(qitem);
+	}
+}
+
 void CentralDB::commitThread()
 {
 	ZTC_LOG("commitThread start\n");
@@ -1188,6 +1232,7 @@ void CentralDB::commitThread()
 
 			if (! qitem.jsonData.is_object()) {
 				ZTC_LOG("commitThread tick: skipping non-object queue item\n");
+				_finishCommit(qitem, NotificationResult::Ok);
 				continue;
 			}
 
@@ -1213,13 +1258,13 @@ void CentralDB::commitThread()
 			}
 			catch (std::exception& e) {
 				ZTC_LOG("ERROR: %s\n", e.what());
-				_requeueFailedCommit(qitem);
+				_finishCommit(qitem, NotificationResult::TransientFailure);
 				continue;
 			}
 
 			if (! c) {
 				ZTC_LOG("Error getting database connection\n");
-				_requeueFailedCommit(qitem);
+				_finishCommit(qitem, NotificationResult::TransientFailure);
 				continue;
 			}
 
@@ -1248,21 +1293,22 @@ void CentralDB::commitThread()
 							target = config["remoteTraceTarget"].get<std::string>();
 						}
 
-						// get network and the frontend it is assigned to
-						// if network does not exist, skip member update
-						pqxx::row nwrow =
-							w.exec("SELECT COUNT(id), frontend FROM networks_ctl WHERE id = $1 GROUP BY frontend",
-								   pqxx::params { networkId })
-								.one_row();
-						int nwcount = nwrow[0].as<int>();
-						std::string frontend = nwrow[1].as<std::string>();
-
-						if (nwcount != 1) {
+						// Get the network's frontend. If the network doesn't exist, this member change
+						// isn't applicable -- it arrived before the network's create (an ordering race)
+						// or after its delete -- so ack-drop it. Treating the absence as a retryable
+						// failure would poison-pill it into repeated redeliveries + a DLQ entry. Use a
+						// plain result + empty() check, not one_row(): a missing network returns zero
+						// rows, which one_row() throws on -> the wrong (retry) path.
+						pqxx::result nwres =
+							w.exec("SELECT frontend FROM networks_ctl WHERE id = $1", pqxx::params { networkId });
+						if (nwres.empty()) {
 							ZTC_LOG("network %s does not exist.  skipping member upsert\n", networkId.c_str());
 							w.abort();
 							_pool->unborrow(c);
+							_finishCommit(qitem, NotificationResult::Ok);
 							continue;
 						}
+						std::string frontend = nwres[0][0].as<std::optional<std::string> >().value_or("");
 
 						pqxx::row mrow =
 							w.exec("SELECT COUNT(device_id) FROM network_memberships_ctl WHERE device_id = $1 "
@@ -1281,6 +1327,9 @@ void CentralDB::commitThread()
 							// the frontend, don't apply the change.
 							w.abort();
 							_pool->unborrow(c);
+							// Change intentionally not applied (network absent, or a non-owning
+							// change source): ack so the message isn't redelivered.
+							_finishCommit(qitem, NotificationResult::Ok);
 							continue;
 						}
 
@@ -1436,6 +1485,9 @@ void CentralDB::commitThread()
 							// the frontend, don't apply the change.
 							w.abort();
 							_pool->unborrow(c);
+							// Change intentionally not applied (network absent, or a non-owning
+							// change source): ack so the message isn't redelivered.
+							_finishCommit(qitem, NotificationResult::Ok);
 							continue;
 						}
 
@@ -1617,11 +1669,12 @@ void CentralDB::commitThread()
 			_pool->unborrow(c);
 			c.reset();
 
-			// Re-queue on a failed DB write so the (already-acked) change isn't lost.
-			// Done after the connection is returned so we never sleep holding one.
-			if (commitFailed) {
-				_requeueFailedCommit(qitem);
-			}
+			// Resolve the item after the DB write, once the connection is returned (so we
+			// never sleep holding one): ack a PubSub-backed change on success / nack on a
+			// transient failure to redeliver, or re-queue an internal write. This replaces
+			// the old unconditional re-queue -- with ack-after-commit the change is no longer
+			// acked at enqueue time, so a failed write is redelivered rather than lost.
+			_finishCommit(qitem, commitFailed ? NotificationResult::TransientFailure : NotificationResult::Ok);
 		}
 	}
 

nonfree/controller/CentralDB.hpp

@@ -9,10 +9,10 @@
 
 #define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 8
 
-// Max times a change is re-queued after a failed DB write before it is dropped.
-// Stopgap durability: the PubSub message was already acked at enqueue time, so a
-// dropped commit would otherwise be lost.  Re-queuing rides out transient DB
-// outages (e.g. an AlloyDB maintenance restart) at the cost of in-memory buffering.
+// Max times an INTERNAL (non-PubSub) write is re-queued after a failed DB write before it is
+// dropped. PubSub-sourced changes are acked/nacked after the commit (ack-after-commit): a failed
+// write nacks so Pub/Sub redelivers, and the subscription's dead-letter policy bounds retries --
+// so this cap governs only internal writes, which have no redelivery and are buffered in memory.
 #define ZT_CENTRAL_CONTROLLER_MAX_COMMIT_RETRIES 100
 
 #include "../../node/Metrics.hpp"
@@ -30,7 +30,7 @@
 namespace ZeroTier {
 struct RedisConfig;
 struct ControllerConfig;
-struct ControllerChangeNotifier;
+class ControllerChangeNotifier;
 class PubSubWriter;
 
 class CentralDB : public DB {
@@ -109,7 +109,7 @@ class CentralDB : public DB {
 	std::string _connString;
 
 	struct _queueItem {
-		_queueItem() : jsonData(), notifyListeners(false), traceContext(), retryCount(0)
+		_queueItem() : jsonData(), notifyListeners(false), traceContext(), retryCount(0), completion()
 		{
 		}
 
@@ -121,13 +121,25 @@ class CentralDB : public DB {
 		bool notifyListeners;
 		std::map<std::string, std::string> traceContext;
 		int retryCount;
+		// Deferred ack for the delivery mechanism (PubSub). Null for internal writes
+		// (no delivery message) and for backends without redelivery semantics. When set,
+		// the commit thread completes it (ack on success, nack on transient failure)
+		// instead of re-queuing -- Pub/Sub's own redelivery replaces _requeueFailedCommit.
+		std::shared_ptr<NotificationCompletion> completion;
 	};
 	BlockingQueue<_queueItem> _commitQueue;
 
 	// Re-queue a change whose DB write failed, with capped backoff, until it
 	// succeeds or ZT_CENTRAL_CONTROLLER_MAX_COMMIT_RETRIES attempts is reached.
+	// Used only for items with no delivery completion (internal writes); items that
+	// carry one are nacked for redelivery instead (see _finishCommit).
 	void _requeueFailedCommit(_queueItem& qitem);
 
+	// Resolve a commit-queue item's outcome exactly once: if it carries a delivery
+	// completion (PubSub), ack on Ok/PermanentFailure or nack on TransientFailure;
+	// otherwise fall back to _requeueFailedCommit on TransientFailure (internal writes).
+	void _finishCommit(_queueItem& qitem, NotificationResult result);
+
 	std::thread _heartbeatThread;
 	std::shared_ptr<NotificationListener> _membersDbWatcher;
 	std::shared_ptr<NotificationListener> _networksDbWatcher;

nonfree/controller/DB.cpp

@@ -467,8 +467,11 @@ void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, boo
 	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
 	auto tracer = provider->GetTracer("db");
 	auto span = tracer->StartSpan("db::_networkChanged");
-	span->SetAttribute("old_network_config", old.dump());
-	span->SetAttribute("network_config", networkConfig.dump());
+	// The full network config is intentionally NOT attached to this span.
+	// old.dump()/networkConfig.dump() serialize the entire (potentially multi-hundred-KB)
+	// config on every network change; carrying that on sampled spans is a large, low-value
+	// memory cost, and it was being built even for un-sampled NoopSpans. The config is always
+	// available in the DB if it's ever needed for debugging.
 	span->SetAttribute("notify_listeners", notifyListeners);
 	auto scope = tracer->WithActiveSpan(span);
 

nonfree/controller/EmbeddedNetworkController.hpp

@@ -35,7 +35,7 @@ class Node;
 struct RedisConfig;
 
 #ifdef ZT1_CENTRAL_CONTROLLER
-class ControllerConfig;
+struct ControllerConfig;
 #endif
 
 class EmbeddedNetworkController

nonfree/controller/NotificationListener.hpp

@@ -5,6 +5,7 @@
 #ifndef NOTIFICATION_LISTENER_HPP
 #define NOTIFICATION_LISTENER_HPP
 
+#include <memory>
 #include <string>
 
 namespace ZeroTier {
@@ -21,6 +22,40 @@ enum class NotificationResult {
 	TransientFailure,	// retryable error (DB unavailable, connection pool exhausted) -> nack to redeliver
 };
 
+/**
+ * Completes the ack/nack for a single notification once its effect has been durably
+ * committed (Ok), determined unprocessable/not-applicable (PermanentFailure -> ack/drop),
+ * or has failed transiently (TransientFailure -> nack/redeliver).
+ *
+ * Lets a delivery mechanism that acks (GCP PubSub) defer the ack until *after* the async
+ * DB commit, so an un-committed change is redelivered rather than lost. Delivery
+ * mechanisms without redelivery semantics (Redis, Postgres LISTEN/NOTIFY) don't create one.
+ * Implementations must be safe to complete exactly once from any thread.
+ */
+class NotificationCompletion {
+  public:
+	virtual ~NotificationCompletion()
+	{
+	}
+	virtual void complete(NotificationResult result) = 0;
+};
+
+/**
+ * Per-thread hand-off used to defer a notification's ack until its asynchronous DB commit
+ * finishes, without threading the completion through every DB/listener signature.
+ *
+ * The delivery callback (e.g. the PubSub subscriber callback) sets the pending completion
+ * around its synchronous onNotification() call. The DB's enqueue path
+ * (CentralDB::save/eraseNetwork/eraseMember) *takes* it -- moving ownership into the
+ * commit-queue item -- when it enqueues async work, and the commit thread completes it after
+ * the write. If the write is a no-op or never reaches the commit queue, the completion is
+ * left in place and the delivery callback completes it synchronously. DB backends that don't
+ * participate (Redis/Postgres, or writes originating off the delivery thread) simply never
+ * find a pending completion and behave as before.
+ */
+void setPendingCompletion(std::shared_ptr<NotificationCompletion> completion);
+std::shared_ptr<NotificationCompletion> takePendingCompletion();
+
 /**
  * Base class for notification listeners
  *

nonfree/controller/PubSubListener.cpp

@@ -23,6 +23,7 @@
 #include <google/cloud/pubsub/admin/subscription_admin_connection.h>
 #include <google/cloud/pubsub/admin/topic_admin_client.h>
 #include <google/cloud/pubsub/message.h>
+#include <google/cloud/pubsub/options.h>
 #include <google/cloud/pubsub/subscriber.h>
 #include <google/cloud/pubsub/subscription.h>
 #include <google/cloud/pubsub/topic.h>
@@ -36,6 +37,36 @@ namespace ZeroTier {
 nlohmann::json toJson(const pbmessages::NetworkChange_Network& nc, pbmessages::NetworkChange_ChangeSource source);
 nlohmann::json toJson(const pbmessages::MemberChange_Member& mc, pbmessages::MemberChange_ChangeSource source);
 
+namespace {
+// Defers a PubSub message's ack/nack until its DB commit completes. Created in the subscriber
+// callback holding the moved-in AckHandler, handed off (as a shared_ptr<NotificationCompletion>)
+// onto the commit-queue item, and completed by the commit thread after the write -- so an
+// un-committed change is redelivered rather than acked-then-lost. Safe to complete exactly once
+// from whichever thread finishes first.
+class PubSubAckCompletion : public NotificationCompletion {
+  public:
+	explicit PubSubAckCompletion(pubsub::AckHandler handler) : _handler(std::move(handler))
+	{
+	}
+	void complete(NotificationResult result) override
+	{
+		if (_done.test_and_set()) {
+			return;	  // ack/nack exactly once
+		}
+		if (result == NotificationResult::TransientFailure) {
+			std::move(_handler).nack();	  // retryable -> redeliver
+		}
+		else {
+			std::move(_handler).ack();	 // Ok / PermanentFailure -> ack (drop on permanent)
+		}
+	}
+
+  private:
+	pubsub::AckHandler _handler;
+	std::atomic_flag _done = ATOMIC_FLAG_INIT;
+};
+}	// namespace
+
 PubSubListener::PubSubListener(std::string controller_id, std::string project, std::string topic)
 	: _controller_id(controller_id)
 	, _project(project)
@@ -60,9 +91,19 @@ PubSubListener::PubSubListener(std::string controller_id, std::string project, s
 		create_gcp_pubsub_subscription_if_needed(_project, _subscription_id, _topic, _controller_id);
 	}
 
-	_subscriber = std::make_shared<pubsub::Subscriber>(
-		pubsub::MakeSubscriberConnection(*_subscription),
-		google::cloud::Options {}.set<google::cloud::OpenTelemetryTracingOption>(true));
+	_subscriber = std::make_shared<pubsub::Subscriber>(pubsub::MakeSubscriberConnection(
+		*_subscription,
+		google::cloud::Options {}
+			.set<google::cloud::OpenTelemetryTracingOption>(true)
+			// Cap the client-side flow-control window. google-cloud-cpp defaults to 1000
+			// outstanding messages / 100 MiB per subscriber; with three controller
+			// subscribers that is up to ~300 MiB of un-acked messages buffered in-process
+			// during a change burst. A smaller window bounds that, and together with
+			// ack-after-commit it also bounds how many uncommitted changes can pile up.
+			// Flow-control options belong on MakeSubscriberConnection (the SubscriberOptionList
+			// consumer), not on the Subscriber constructor.
+			.set<pubsub::MaxOutstandingMessagesOption>(200)
+			.set<pubsub::MaxOutstandingBytesOption>(16 * 1024 * 1024)));
 
 	_run = true;
 	_subscriberThread = std::thread(&PubSubListener::subscribe, this);
@@ -104,6 +145,11 @@ void PubSubListener::subscribe()
 
 			auto session = _subscriber->Subscribe([this](pubsub::Message const& m, pubsub::AckHandler h) {
 				_lastMessageTime.store(std::chrono::steady_clock::now());
+				// Defer the ack until the change is durably committed: hand ownership of the
+				// AckHandler to a completion that the DB's commit thread resolves. If the write is
+				// a no-op, never reaches the commit queue, or is handled synchronously (SSO), we
+				// complete it ourselves below with onNotification's reported outcome.
+				setPendingCompletion(std::make_shared<PubSubAckCompletion>(std::move(h)));
 				// Default to transient: if anything throws before onNotification reports an
 				// outcome, redeliver rather than silently dropping the change.
 				NotificationResult result = NotificationResult::TransientFailure;
@@ -168,11 +214,27 @@ void PubSubListener::subscribe()
 					result = NotificationResult::PermanentFailure;
 				}
 
-				if (result == NotificationResult::TransientFailure) {
-					std::move(h).nack();
-				}
-				else {
-					std::move(h).ack();
+				// If onNotification enqueued async DB work it took the pending completion (the
+				// commit thread acks/nacks after the write). Otherwise -- a no-op, a
+				// parse/validation failure, or a synchronous handler -- complete it now with the
+				// outcome onNotification reported.
+				if (auto completion = takePendingCompletion()) {
+					// Guard the ack/nack: an exception escaping this subscriber callback can tear
+					// down the pull session. On a throwing path ack()/nack() has already consumed
+					// the handler, so there's nothing to retry here -- just don't let it propagate.
+					try {
+						completion->complete(result);
+					}
+					catch (const std::exception& e) {
+						ZTC_LOG(
+							"PubSubListener: ack/nack failed (message_id=%s): %s\n", m.message_id().c_str(),
+							e.what());
+					}
+					catch (...) {
+						ZTC_LOG(
+							"PubSubListener: ack/nack failed (message_id=%s); unknown exception\n",
+							m.message_id().c_str());
+					}
 				}
 				return true;
 			});

nonfree/controller/PubSubListener.hpp

@@ -30,7 +30,7 @@ class PubSubListener : public NotificationListener {
 	PubSubListener(std::string controller_id, std::string project, std::string topic);
 	virtual ~PubSubListener();
 
-	virtual NotificationResult onNotification(const std::string& payload) = 0;
+	virtual NotificationResult onNotification(const std::string& payload) override = 0;
 
 	// Stop and join the subscriber thread. Must be called from the most-derived
 	// destructor (before its members are torn down) so an in-flight callback can't