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

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

ext/central-controller-docker/main.sh

@@ -179,7 +179,7 @@ echo "{
         ${BIGTABLE_CONF}
         ${PUBSUB_CONF}
     }
-}    
+}
 " > /var/lib/zerotier-one/local.conf
 
 if [ -n "$DB_SERVER_CA" ]; then
@@ -198,12 +198,12 @@ else
 fi
 
 
-echo "Migrating database (if needed)..."
-if [ -n "$DB_SERVER_CA" ]; then
-    /usr/local/bin/migrate -source file:///migrations -database "postgres://$ZT_DB_USER:$ZT_DB_PASSWORD@$ZT_DB_HOST:$ZT_DB_PORT/$ZT_DB_NAME?x-migrations-table=controller_migrations&sslmode=verify-full&sslrootcert=$DB_SERVER_CA&sslcert=$DB_CLIENT_CERT&sslkey=$DB_CLIENT_KEY" up  
-else 
-    /usr/local/bin/migrate -source file:///migrations -database "postgres://$ZT_DB_USER:$ZT_DB_PASSWORD@$ZT_DB_HOST:$ZT_DB_PORT/$ZT_DB_NAME?x-migrations-table=controller_migrations&sslmode=disable" up
-fi
+# echo "Migrating database (if needed)..."
+# if [ -n "$DB_SERVER_CA" ]; then
+#     /usr/local/bin/migrate -source file:///migrations -database "postgres://$ZT_DB_USER:$ZT_DB_PASSWORD@$ZT_DB_HOST:$ZT_DB_PORT/$ZT_DB_NAME?x-migrations-table=controller_migrations&sslmode=verify-full&sslrootcert=$DB_SERVER_CA&sslcert=$DB_CLIENT_CERT&sslkey=$DB_CLIENT_KEY" up
+# else
+#     /usr/local/bin/migrate -source file:///migrations -database "postgres://$ZT_DB_USER:$ZT_DB_PASSWORD@$ZT_DB_HOST:$ZT_DB_PORT/$ZT_DB_NAME?x-migrations-table=controller_migrations&sslmode=disable" up
+# fi
 
 if [ -n "$ZT_TEMPORAL_HOST" ] && [ -n "$ZT_TEMPORAL_PORT" ]; then
     echo "waiting for temporal..."

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);
 		}
 	}
 
@@ -1791,6 +1844,10 @@ nlohmann::json CentralDB::_getNetworkMember(pqxx::work& tx, const std::string ne
 		out["creationTime"] = creation_time.value_or(0);
 		out["lastAuthorizedTime"] = last_authorized_time.value_or(0);
 		out["lastDeauthorizedTime"] = last_deauthorized_time.value_or(0);
+		// Not stored in the members table; emit empty defaults so oldMember matches
+		// the shape memberFromJson (PubSubWriter.cpp) reads.
+		out["lastAuthorizedCredential"] = "";
+		out["lastAuthorizedCredentialType"] = "";
 		out["noAutoAssignIps"] = no_auto_assign_ips.value_or(false);
 		out["remoteTraceLevel"] = remote_trace_level.value_or(0);
 		out["remoteTraceTarget"] = remote_trace_target.value_or(nullptr);

nonfree/controller/CentralDB.hpp

@@ -7,12 +7,12 @@
 #ifndef ZT_CONTROLLER_CENTRAL_DB_HPP
 #define ZT_CONTROLLER_CENTRAL_DB_HPP
 
-#define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4
+#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.cpp

@@ -51,7 +51,10 @@ using json = nlohmann::json;
 // Cap on the request queue for controller-initiated re-requests (pushes), which fan
 // out to every online member on a network change. Beyond this they're dropped (the
 // member picks up the change on its next poll) to bound memory under churn.
-#define ZT_CONTROLLER_MAX_QUEUED_REREQUESTS 16384
+// Hard safety backstop on the request-queue depth. Per-member de-dup (see request()) keeps the
+// real depth ~= the online-member count; this only bounds pathological overflow. ~1.3 KB/entry,
+// so 2097152 ~= 2.6 GB worst case (well within the pod memory limit).
+#define ZT_CONTROLLER_MAX_QUEUED_REREQUESTS 2097152
 
 namespace ZeroTier {
 
@@ -740,25 +743,48 @@ void EmbeddedNetworkController::request(
 		}
 		ms.lastRequestTime = now;
 	}
-	else {
-		// Controller-initiated re-request (push). These fan out to every online member
-		// on a network change, so cap the backlog and drop on overflow rather than
-		// growing the queue without bound — the member will still pick up the change on
-		// its next periodic poll. Member-initiated requests (requestPacketId != 0) are
-		// always queued; they're already rate-limited per member above.
-		if (_queue.size() >= ZT_CONTROLLER_MAX_QUEUED_REREQUESTS) {
-			return;
+
+	// Hard backstop on total depth (see the constant). Both a controller-initiated re-request
+	// (a network-change fan-out to every online member) and a member-initiated reconnect herd
+	// can enqueue faster than the worker threads drain; over the backstop we drop (the member
+	// re-requests on its next poll; a controller push re-applies later).
+	if (_queue.size() >= ZT_CONTROLLER_MAX_QUEUED_REREQUESTS) {
+		return;
+	}
+
+	// De-dup: keep at most one pending request per (network, member) in the queue. Member
+	// requests were previously only per-member rate-limited (1s), so a reconnect herd across
+	// ~24k members each re-requesting ~1x/sec while the backlog drained stacked ~20+ duplicate
+	// _RQEntry per member -- ~500k entries (~600 MB) and an OOM. De-duping bounds the real depth
+	// to roughly the online-member count; nothing is lost because the queued request reads fresh
+	// config when a worker processes it, and the key is cleared on dequeue so a change arriving
+	// mid-process still re-queues.
+	const _MemberStatusKey rqKey(nwid, identity.address().toInt());
+	{
+		std::lock_guard<std::mutex> l(_pendingRequests_l);
+		if (! _pendingRequests.insert(rqKey).second) {
+			return;	  // a request for this member is already queued
 		}
 	}
 
-	_RQEntry* qe = new _RQEntry;
-	qe->nwid = nwid;
-	qe->requestPacketId = requestPacketId;
-	qe->fromAddr = fromAddr;
-	qe->identity = identity;
-	qe->metaData = metaData;
-	qe->type = _RQEntry::RQENTRY_TYPE_REQUEST;
-	_queue.post(qe);
+	// The member is now marked pending, and that mark is only cleared when a worker dequeues the
+	// entry. If we fail to enqueue after this point (allocation/copy/post throwing), erase the
+	// mark so the member isn't permanently deduped away and starved until the process restarts.
+	try {
+		std::unique_ptr<_RQEntry> qe(new _RQEntry);
+		qe->nwid = nwid;
+		qe->requestPacketId = requestPacketId;
+		qe->fromAddr = fromAddr;
+		qe->identity = identity;
+		qe->metaData = metaData;
+		qe->type = _RQEntry::RQENTRY_TYPE_REQUEST;
+		_queue.post(qe.get());
+		qe.release();	 // ownership handed to _queue; the worker deletes it
+	}
+	catch (...) {
+		std::lock_guard<std::mutex> l(_pendingRequests_l);
+		_pendingRequests.erase(rqKey);
+	}
 }
 
 std::string EmbeddedNetworkController::networkUpdateFromPostData(uint64_t networkID, const std::string& body)
@@ -2490,7 +2516,12 @@ void EmbeddedNetworkController::_startThreads()
 	if (! _threads.empty()) {
 		return;
 	}
+#ifdef ZT1_CENTRAL_CONTROLLER
+    // Ensure central controllers have at least 8 threads running to process network config requests
+    const long hwc = std::max((long)std::thread::hardware_concurrency(), (long)8);
+#else
 	const long hwc = std::max((long)std::thread::hardware_concurrency(), (long)1);
+#endif
 	for (long t = 0; t < hwc; ++t) {
 		_threads.emplace_back([this, t]() {
 			Metrics::network_config_request_threads++;
@@ -2507,6 +2538,13 @@ void EmbeddedNetworkController::_startThreads()
 				else if (timedWaitResult == BlockingQueue<_RQEntry*>::OK) {
 					idleIterations = 0;
 					if (qe) {
+						// Clear the de-dup mark now that this entry is dequeued, so a fresh request
+						// for the same member (e.g. a config change arriving while we process it) can
+						// queue again instead of being deduped away.
+						{
+							std::lock_guard<std::mutex> l(_pendingRequests_l);
+							_pendingRequests.erase(_MemberStatusKey(qe->nwid, qe->identity.address().toInt()));
+						}
 						try {
 							_request(qe->nwid, qe->fromAddr, qe->requestPacketId, qe->identity, qe->metaData);
 							processedCount++;

nonfree/controller/EmbeddedNetworkController.hpp

@@ -26,6 +26,7 @@
 #include <string>
 #include <thread>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 namespace ZeroTier {
@@ -34,7 +35,7 @@ class Node;
 struct RedisConfig;
 
 #ifdef ZT1_CENTRAL_CONTROLLER
-class ControllerConfig;
+struct ControllerConfig;
 #endif
 
 class EmbeddedNetworkController
@@ -157,6 +158,12 @@ class EmbeddedNetworkController
 	std::unordered_map<_MemberStatusKey, _MemberStatus, _MemberStatusHash> _memberStatus;
 	std::mutex _memberStatus_l;
 
+	// At-most-one queued config request per (network, member): a reconnect herd (or a member
+	// re-requesting while an earlier request is still queued) would otherwise stack duplicate
+	// _RQEntry (each ~1 KB). Inserted in request(), erased when a worker dequeues the entry.
+	std::unordered_set<_MemberStatusKey, _MemberStatusHash> _pendingRequests;
+	std::mutex _pendingRequests_l;
+
 	std::set<std::pair<int64_t, _MemberStatusKey> > _expiringSoon;
 	std::mutex _expiringSoon_l;
 

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

nonfree/controller/PubSubWriter.cpp

@@ -29,6 +29,19 @@ networkChangeFromJson(std::string controllerID, const nlohmann::json& oldNetwork
 pbmessages::MemberChange*
 memberChangeFromJson(std::string controllerID, const nlohmann::json& oldMember, const nlohmann::json& newMember);
 
+namespace {
+// Safe const field lookup. Returns a reference to j[key], or to a shared null
+// json when key is absent (or j is not an object). Avoids the undefined
+// behavior of nlohmann's const operator[] on a missing key, whose only guard
+// (JSON_ASSERT) is compiled out in release/NDEBUG builds.
+const nlohmann::json& jfield(const nlohmann::json& j, const char* key)
+{
+	static const nlohmann::json kNull;
+	auto it = j.find(key);	 // find() is safe even when j is not an object
+	return (it != j.end()) ? *it : kNull;
+}
+}	// namespace
+
 PubSubWriter::PubSubWriter(std::string project, std::string topic, std::string controller_id)
 	: _controller_id(controller_id)
 	, _project(project)
@@ -228,37 +241,37 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 
 	pbmessages::NetworkChange_Network* n = new pbmessages::NetworkChange_Network();
 	try {
-		n->set_network_id(OSUtils::jsonString(j["id"], ""));
-		n->set_name(OSUtils::jsonString(j["name"], ""));
+		n->set_network_id(OSUtils::jsonString(jfield(j, "id"), ""));
+		n->set_name(OSUtils::jsonString(jfield(j, "name"), ""));
 		n->set_capabilities(OSUtils::jsonDump(j.value("capabilities", "[]"), -1));
-		n->set_creation_time(OSUtils::jsonInt(j["creationTime"], 0));
-		n->set_enable_broadcast(OSUtils::jsonBool(j["enableBroadcast"], false));
+		n->set_creation_time(OSUtils::jsonInt(jfield(j, "creationTime"), 0));
+		n->set_enable_broadcast(OSUtils::jsonBool(jfield(j, "enableBroadcast"), false));
 
-		for (const auto& p : j["ipAssignmentPools"]) {
+		for (const auto& p : jfield(j, "ipAssignmentPools")) {
 			if (p.is_object()) {
 				auto pool = n->add_assignment_pools();
-				pool->set_start_ip(OSUtils::jsonString(p["ipRangeStart"], ""));
-				pool->set_end_ip(OSUtils::jsonString(p["ipRangeEnd"], ""));
+				pool->set_start_ip(OSUtils::jsonString(jfield(p, "ipRangeStart"), ""));
+				pool->set_end_ip(OSUtils::jsonString(jfield(p, "ipRangeEnd"), ""));
 			}
 		}
 
-		n->set_mtu(OSUtils::jsonInt(j["mtu"], 2800));
-		n->set_multicast_limit(OSUtils::jsonInt(j["multicastLimit"], 32));
-		n->set_is_private(OSUtils::jsonBool(j["private"], true));
-		n->set_remote_trace_level(OSUtils::jsonInt(j["remoteTraceLevel"], 0));
-		n->set_remote_trace_target(OSUtils::jsonString(j["remoteTraceTarget"], ""));
-		n->set_revision(OSUtils::jsonInt(j["revision"], 0));
+		n->set_mtu(OSUtils::jsonInt(jfield(j, "mtu"), 2800));
+		n->set_multicast_limit(OSUtils::jsonInt(jfield(j, "multicastLimit"), 32));
+		n->set_is_private(OSUtils::jsonBool(jfield(j, "private"), true));
+		n->set_remote_trace_level(OSUtils::jsonInt(jfield(j, "remoteTraceLevel"), 0));
+		n->set_remote_trace_target(OSUtils::jsonString(jfield(j, "remoteTraceTarget"), ""));
+		n->set_revision(OSUtils::jsonInt(jfield(j, "revision"), 0));
 
-		for (const auto& p : j["routes"]) {
+		for (const auto& p : jfield(j, "routes")) {
 			if (p.is_object()) {
 				auto r = n->add_routes();
-				r->set_target(OSUtils::jsonString(p["target"], ""));
-				r->set_via(OSUtils::jsonString(p["via"], ""));
+				r->set_target(OSUtils::jsonString(jfield(p, "target"), ""));
+				r->set_via(OSUtils::jsonString(jfield(p, "via"), ""));
 			}
 		}
 		std::string rules;
-		if (j["rules"].is_array()) {
-			rules = OSUtils::jsonDump(j["rules"], -1);
+		if (jfield(j, "rules").is_array()) {
+			rules = OSUtils::jsonDump(jfield(j, "rules"), -1);
 		}
 		else {
 			rules = "[]";
@@ -266,8 +279,8 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 		n->set_rules(rules);
 
 		std::string tags;
-		if (j["tags"].is_array()) {
-			tags = OSUtils::jsonDump(j["tags"], -1);
+		if (jfield(j, "tags").is_array()) {
+			tags = OSUtils::jsonDump(jfield(j, "tags"), -1);
 		}
 		else {
 			tags = "[]";
@@ -275,22 +288,22 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 		n->set_tags(tags);
 
 		pbmessages::NetworkChange_IPV4AssignMode* v4am = new pbmessages::NetworkChange_IPV4AssignMode();
-		if (j["v4AssignMode"].is_object()) {
-			nlohmann::json am = j["v4AssignMode"];
+		if (jfield(j, "v4AssignMode").is_object()) {
+			nlohmann::json am = jfield(j, "v4AssignMode");
 			v4am->set_zt(OSUtils::jsonBool(am["zt"], false));
 		}
 		n->set_allocated_ipv4_assign_mode(v4am);
 
 		pbmessages::NetworkChange_IPV6AssignMode* v6am = new pbmessages::NetworkChange_IPV6AssignMode();
-		if (j["v6AssignMode"].is_object()) {
-			nlohmann::json am = j["v6AssignMode"];
+		if (jfield(j, "v6AssignMode").is_object()) {
+			nlohmann::json am = jfield(j, "v6AssignMode");
 			v6am->set_zt(OSUtils::jsonBool(am["zt"], false));
 			v6am->set_six_plane(OSUtils::jsonBool(am["6plane"], false));
 			v6am->set_rfc4193(OSUtils::jsonBool(am["rfc4193"], false));
 		}
 		n->set_allocated_ipv6_assign_mode(v6am);
 
-		nlohmann::json jdns = j["dns"];
+		nlohmann::json jdns = jfield(j, "dns");
 		if (jdns.is_object()) {
 			pbmessages::NetworkChange_DNS* dns = new pbmessages::NetworkChange_DNS();
 			dns->set_domain(jdns.value("domain", ""));
@@ -303,14 +316,14 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 			n->set_allocated_dns(dns);
 		}
 
-		n->set_sso_enabled(OSUtils::jsonBool(j["ssoEnabled"], false));
-		nlohmann::json ssocfg = j["ssoConfig"];
+		n->set_sso_enabled(OSUtils::jsonBool(jfield(j, "ssoEnabled"), false));
+		nlohmann::json ssocfg = jfield(j, "ssoConfig");
 		if (ssocfg.is_object()) {
 			n->set_sso_client_id(OSUtils::jsonString(ssocfg["ssoClientId"], ""));
 			n->set_sso_linked_id(OSUtils::jsonString(ssocfg["ssoLinkedId"], ""));
 		}
 
-		n->set_rules_source(OSUtils::jsonString(j["rulesSource"], ""));
+		n->set_rules_source(OSUtils::jsonString(jfield(j, "rulesSource"), ""));
 	}
 	catch (const std::exception& e) {
 		ZTC_LOG("Exception parsing network JSON: %s\n", e.what());
@@ -346,12 +359,12 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 
 	pbmessages::MemberChange_Member* m = new pbmessages::MemberChange_Member();
 	try {
-		m->set_network_id(OSUtils::jsonString(j["nwid"], ""));
-		m->set_device_id(OSUtils::jsonString(j["id"], ""));
-		m->set_identity(OSUtils::jsonString(j["identity"], ""));
-		m->set_authorized(OSUtils::jsonBool(j["authorized"], false));
-		if (j["ipAssignments"].is_array()) {
-			for (const auto& addr : j["ipAssignments"]) {
+		m->set_network_id(OSUtils::jsonString(jfield(j, "nwid"), ""));
+		m->set_device_id(OSUtils::jsonString(jfield(j, "id"), ""));
+		m->set_identity(OSUtils::jsonString(jfield(j, "identity"), ""));
+		m->set_authorized(OSUtils::jsonBool(jfield(j, "authorized"), false));
+		if (jfield(j, "ipAssignments").is_array()) {
+			for (const auto& addr : jfield(j, "ipAssignments")) {
 				if (addr.is_string()) {
 					auto a = m->add_ip_assignments();
 					std::string address = addr.get<std::string>();
@@ -359,9 +372,9 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 				}
 			}
 		}
-		m->set_active_bridge(OSUtils::jsonBool(j["activeBridge"], false));
-		if (j["tags"].is_array()) {
-			nlohmann::json tags = j["tags"];
+		m->set_active_bridge(OSUtils::jsonBool(jfield(j, "activeBridge"), false));
+		if (jfield(j, "tags").is_array()) {
+			nlohmann::json tags = jfield(j, "tags");
 			std::string tagsStr = OSUtils::jsonDump(tags, -1);
 			m->set_tags(tagsStr);
 		}
@@ -370,8 +383,8 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 			std::string tagsStr = OSUtils::jsonDump(tags, -1);
 			m->set_tags(tagsStr);
 		}
-		if (j["capabilities"].is_array()) {
-			nlohmann::json caps = j["capabilities"];
+		if (jfield(j, "capabilities").is_array()) {
+			nlohmann::json caps = jfield(j, "capabilities");
 			std::string capsStr = OSUtils::jsonDump(caps, -1);
 			m->set_capabilities(capsStr);
 		}
@@ -380,21 +393,21 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 			std::string capsStr = OSUtils::jsonDump(caps, -1);
 			m->set_capabilities(capsStr);
 		}
-		m->set_creation_time(OSUtils::jsonInt(j["creationTime"], 0));
-		m->set_no_auto_assign_ips(OSUtils::jsonBool(j["noAutoAssignIps"], false));
-		m->set_revision(OSUtils::jsonInt(j["revision"], 0));
-		m->set_last_authorized_time(OSUtils::jsonInt(j["lastAuthorizedTime"], 0));
-		m->set_last_deauthorized_time(OSUtils::jsonInt(j["lastDeauthorizedTime"], 0));
-		m->set_last_authorized_credential_type(OSUtils::jsonString(j["lastAuthorizedCredentialType"], ""));
-		m->set_last_authorized_credential(OSUtils::jsonString(j["lastAuthorizedCredential"], ""));
-		m->set_version_major(OSUtils::jsonInt(j["vMajor"], 0));
-		m->set_version_minor(OSUtils::jsonInt(j["vMinor"], 0));
-		m->set_version_rev(OSUtils::jsonInt(j["vRev"], 0));
-		m->set_version_protocol(OSUtils::jsonInt(j["vProto"], 0));
-		m->set_remote_trace_level(OSUtils::jsonInt(j["remoteTraceLevel"], 0));
-		m->set_remote_trace_target(OSUtils::jsonString(j["remoteTraceTarget"], ""));
-		m->set_sso_exempt(OSUtils::jsonBool(j["ssoExempt"], false));
-		m->set_auth_expiry_time(OSUtils::jsonInt(j["authenticationExpiryTime"], 0));
+		m->set_creation_time(OSUtils::jsonInt(jfield(j, "creationTime"), 0));
+		m->set_no_auto_assign_ips(OSUtils::jsonBool(jfield(j, "noAutoAssignIps"), false));
+		m->set_revision(OSUtils::jsonInt(jfield(j, "revision"), 0));
+		m->set_last_authorized_time(OSUtils::jsonInt(jfield(j, "lastAuthorizedTime"), 0));
+		m->set_last_deauthorized_time(OSUtils::jsonInt(jfield(j, "lastDeauthorizedTime"), 0));
+		m->set_last_authorized_credential_type(OSUtils::jsonString(jfield(j, "lastAuthorizedCredentialType"), ""));
+		m->set_last_authorized_credential(OSUtils::jsonString(jfield(j, "lastAuthorizedCredential"), ""));
+		m->set_version_major(OSUtils::jsonInt(jfield(j, "vMajor"), 0));
+		m->set_version_minor(OSUtils::jsonInt(jfield(j, "vMinor"), 0));
+		m->set_version_rev(OSUtils::jsonInt(jfield(j, "vRev"), 0));
+		m->set_version_protocol(OSUtils::jsonInt(jfield(j, "vProto"), 0));
+		m->set_remote_trace_level(OSUtils::jsonInt(jfield(j, "remoteTraceLevel"), 0));
+		m->set_remote_trace_target(OSUtils::jsonString(jfield(j, "remoteTraceTarget"), ""));
+		m->set_sso_exempt(OSUtils::jsonBool(jfield(j, "ssoExempt"), false));
+		m->set_auth_expiry_time(OSUtils::jsonInt(jfield(j, "authenticationExpiryTime"), 0));
 	}
 	catch (const std::exception& e) {
 		ZTC_LOG("Exception parsing member JSON: %s\n", e.what());