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

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

README.md

@@ -1,19 +1,4 @@
-<p align="center">
-  <img src="artwork/ZeroTierIcon512x512.png" alt="ZeroTier" width="120">
-</p>
-
-<h1 align="center">ZeroTier One</h1>
-
-<p align="center"><em>Global Area Networking</em></p>
-
-<p align="center">
-  <a href="https://github.com/zerotier/ZeroTierOne/actions/workflows/build.yml"><img src="https://github.com/zerotier/ZeroTierOne/actions/workflows/build.yml/badge.svg" alt="Build"></a>
-  <a href="https://github.com/zerotier/ZeroTierOne/actions/workflows/validate.yml"><img src="https://github.com/zerotier/ZeroTierOne/actions/workflows/validate.yml/badge.svg" alt="Validate"></a>
-  <a href="https://github.com/zerotier/ZeroTierOne/releases"><img src="https://img.shields.io/github/v/release/zerotier/ZeroTierOne?label=release&color=1a3d8f" alt="Latest release"></a>
-  <a href="#license"><img src="https://img.shields.io/badge/license-MPL--2.0%20%2B%20source--available-blue" alt="License"></a>
-  <a href="https://docs.zerotier.com"><img src="https://img.shields.io/badge/docs-docs.zerotier.com-1a3d8f" alt="Documentation"></a>
-  <img src="https://img.shields.io/badge/platforms-Linux%20%7C%20macOS%20%7C%20Windows%20%7C%20BSD%20%7C%20Android%20%7C%20iOS-lightgrey" alt="Platforms">
-</p>
+# ZeroTier - Global Area Networking
 
 ## Quick Links
 
@@ -23,32 +8,24 @@
 * [Downloads](https://www.zerotier.com/download/)
 * [Service API Reference](service/README.md)
 * [Network Controller](nonfree/controller/README.md)
-* [Commercial Support](https://docs.zerotier.com/support/)
+* [Commercial Support](https://www.zerotier.com/contact)
 * [License Information](#license)
 
 ## About
 
-ZeroTier One is modern, identity-first software-defined networking for distributed infrastructure. It connects devices, servers, VMs, containers, and applications into one secure virtual network — no matter what sits between them — so they can communicate as if they all reside on the same physical LAN, without specialized hardware or complex VPN configuration.
+ZeroTier is a smart programmable Ethernet switch for planet Earth. It allows all networked devices, VMs, containers, and applications to communicate as if they all reside in the same physical data center or cloud region.
 
-### How It Works
+This is accomplished by combining a cryptographically addressed and secure peer-to-peer network (termed VL1) with an Ethernet emulation layer somewhat similar to VXLAN (termed VL2). Our VL2 Ethernet virtualization layer includes advanced enterprise SDN features like fine grained access control rules for network micro-segmentation and security monitoring.
 
-Instead of relying on physical topology, every device joins a network using its own cryptographic identity. This is built on a secure peer-to-peer transport layer (VL1) combined with an Ethernet virtualization layer similar to VXLAN (VL2), which gives you fine-grained, capability-based access control rules for network micro-segmentation and security monitoring. Access policy is centrally defined and globally enforced, so networks stay resilient and zero-trust by default across public clouds, private data centers, on-prem environments, remote devices, and edge systems.
+All ZeroTier traffic is encrypted end-to-end using secret keys that only you control. Most traffic flows peer-to-peer, though we offer free (but slow) relaying for users who cannot establish peer-to-peer connections.
 
-### Security
-
-All ZeroTier traffic is authenticated and encrypted end-to-end using keys that only you control. Connections are established peer-to-peer whenever possible, with free (but slower) relaying available for devices that can't establish a direct path.
-
-## Platforms
-
-ZeroTier One runs on Linux, macOS, Windows, and BSD. Apps for Android and iOS are available for free in the Google Play and Apple App Store.
-
-## Building and Running
+Apps for Android and iOS are available for free in the Google Play and Apple app stores.
 
 For repository layout, build instructions, platform requirements, and information about running ZeroTier, see [build.md](build.md).
 
 ## License
 
-See [LICENSE-MPL.txt](LICENSE-MPL.txt) for all code in node/, osdep/, service/, and everywhere else except ext/ and nonfree/.
+See [LICENSE-MPL.txt](LICENSE-MPL.txt) for all code in node/, osdep/. service/, and everywhere else except ext/ and nonfree/.
 
 See [nonfree/LICENSE.md](nonfree/LICENSE.md) for all non-free ("source available") portions of this repository.
 

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());

rustybits/Cargo.toml

@@ -24,6 +24,12 @@ time = { version = "~0.3", features = ["formatting"] }
 bytes = "1.3"
 thiserror = "1"
 
+[dev-dependencies]
+testcontainers = { version = "0.24", features = ["blocking"] }
+testcontainers-modules = { version = "0.12.1", features = [
+    "google_cloud_sdk_emulators",
+] }
+
 [build-dependencies]
 cbindgen = "0.29"