index fixups for the prod DB

Reduce spans emitted per controller request without touching the head
sample rate. Removes 20 StartSpan sites (97 -> 77) that traced the same
logical op at multiple layers:

- DB.cpp: drop all db:: read-helper/leaf spans; the db_mirror_set::
  wrapper survives as the single span per read op (also removes the
  latent member_id-from-networkId attribute bug).
- CentralDB.cpp: drop inner write-op spans (save/+children, nodeIsOnline,
  eraseNetwork, eraseMember); keep getSSOAuthInfo (errors) and all
  background-thread spans (commitThread, onlineNotificationThread, init).
- EmbeddedNetworkController.cpp: drop per-rule parseRule span and the
  no-op _startThreads span; keep request/_request roots.

Sampler, trace-context propagation, commit/Bigtable-write spans, and all
business logic are unchanged.

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..."

ext/central-controller-docker/migrations/0006_oidc_config.up.sql

@@ -1,4 +1,9 @@
 ALTER TABLE sso_expiry RENAME COLUMN member_id TO device_id;
+-- NOTE (2026-08-03): sso_expiry_network_member_ix is dropped again in 0008. It looks
+-- correct but could never serve the ON DELETE CASCADE from network_memberships_ctl --
+-- sso_expiry's columns are bpchar against a varchar parent, so the RI trigger compares
+-- network_id::text and no bare-column index can match it. Do not re-add a bare-column
+-- index here expecting it to help; see 0008_sso_expiry_indexes.up.sql for the full story.
 CREATE INDEX IF NOT EXISTS sso_expiry_network_member_ix ON public.sso_expiry (network_id, device_id);
 
 CREATE TABLE IF NOT EXISTS oidc_config (

ext/central-controller-docker/migrations/0008_sso_expiry_indexes.down.sql

@@ -0,0 +1,15 @@
+-- Reverse of 0008: drop the two new indexes and restore sso_expiry_network_member_ix
+-- exactly as 0006 created it.
+--
+-- WARNING: this reinstates the defect 0008 fixed. With sso_expiry_ri_ix gone, the
+-- ON DELETE CASCADE from network_memberships_ctl falls back to a full sequential scan of
+-- sso_expiry for every member deleted; with sso_expiry_lookup_ix gone the netconf-path
+-- nonce lookups go back to heap-fetching every accumulated row per member; and with
+-- sso_expiry_creation_time_ix gone the retention pruner full-scans the table every run.
+-- See the header of 0008_sso_expiry_indexes.up.sql for the measurements.
+
+DROP INDEX IF EXISTS sso_expiry_creation_time_ix;
+DROP INDEX IF EXISTS sso_expiry_lookup_ix;
+DROP INDEX IF EXISTS sso_expiry_ri_ix;
+
+CREATE INDEX IF NOT EXISTS sso_expiry_network_member_ix ON public.sso_expiry (network_id, device_id);

ext/central-controller-docker/migrations/0008_sso_expiry_indexes.up.sql

@@ -0,0 +1,71 @@
+-- sso_expiry index overhaul (2026-08-03).
+--
+-- WHY: sso_expiry.network_id / device_id are CHARACTER(16) / CHARACTER(10) (bpchar), while
+-- the parent columns they reference -- network_memberships_ctl.network_id / device_id --
+-- are character varying(22). There is no varchar = bpchar operator, so when the FK in 0005
+-- was created PostgreSQL coerced both sides to text and stored texteq as the constraint's
+-- comparison operator. The ON DELETE CASCADE trigger consequently emits:
+--
+--     DELETE FROM ONLY public.sso_expiry x
+--     WHERE $1::pg_catalog.text OPERATOR(pg_catalog.=) x.network_id::pg_catalog.text
+--       AND $2::pg_catalog.text OPERATOR(pg_catalog.=) x.device_id::pg_catalog.text
+--
+-- bpchar->text is a real function call (it strips trailing blanks), not a binary-coercible
+-- relabel, so a btree index on the bare columns can NEVER match that qual. This is why
+-- sso_expiry_network_member_ix (added in 0006, on exactly the right two columns) was never
+-- once used by the cascade: every single member delete seq-scanned the entire table.
+--
+-- Measured in prod 2026-08-01: 64,974 buffers / 1.33 s per deleted member, at idle. A user
+-- bulk-deleting ~2,100 members from one network pinned the 4-vCPU AlloyDB instance at 100%
+-- CPU for 27 minutes and paged on controller_db_commit_latency_ms.
+--
+-- Note the asymmetry: the reverse RI direction (checking the parent exists on INSERT into
+-- sso_expiry) stays fast at ~17 buffers, because varchar->text IS binary-coercible and the
+-- planner strips it as a RelabelType. Same operator, same cast syntax, opposite outcomes.
+--
+-- WHAT:
+--   sso_expiry_ri_ix     -- built on the cast expressions so the cascade can match it.
+--                           Prod: 61,268 -> ~15 buffers, 1,450 ms -> 0.4 ms per delete.
+--   sso_expiry_lookup_ix -- adds nonce_expiration as a third key column so the netconf-path
+--                           nonce lookups (CentralDB.cpp, "SELECT nonce FROM sso_expiry
+--                           WHERE network_id = $1 AND device_id = $2 AND ... <= ...") can
+--                           range-seek past expired rows instead of heap-fetching every
+--                           accumulated row for that member. A member that never completes
+--                           SSO accrues one row per 5 minutes and rows are retained ~30
+--                           days (sso_expiry doubles as the SSO log), so those groups run
+--                           to thousands of rows.
+--                           Prod: 4,850 -> 5 buffers, 16 ms -> 0.03 ms per lookup, which
+--                           returned ~1.2 of the 4 vCPUs.
+--   sso_expiry_creation_time_ix
+--                        -- the retention pruner in central-v2/apps/controller-manager/
+--                           core/controller/ssocleanup.go ("DELETE FROM sso_expiry WHERE
+--                           creation_time < NOW() - INTERVAL '32 days'") had no index on
+--                           creation_time and full-scanned the table on every run, roughly
+--                           every 5 minutes, forever.
+--                           Prod: 58,000 -> ~900 buffers, 880 ms -> 1.0 ms per run.
+--   sso_expiry_network_member_ix is dropped: it is a strict prefix of sso_expiry_lookup_ix,
+--                           so it serves nothing the new index does not, while costing
+--                           maintenance on every insert and every pruner delete.
+--
+-- LOCKING: prod already has all three changes, applied by hand on 2026-08-03 using
+-- CREATE INDEX CONCURRENTLY / DROP INDEX CONCURRENTLY, so nothing blocked. This file
+-- deliberately uses plain, non-concurrent DDL: golang-migrate sends the whole file as a
+-- single Exec, which PostgreSQL runs as an implicit transaction block, and
+-- CREATE INDEX CONCURRENTLY cannot run inside a transaction block. That is the right
+-- trade-off for the databases this file will actually run against -- dev, staging, and
+-- freshly provisioned controllers -- where sso_expiry is small or empty.
+--
+-- Do NOT run this file against a large populated database. CREATE INDEX takes a SHARE lock
+-- (blocking writes to sso_expiry for the duration of each build) and DROP INDEX takes
+-- ACCESS EXCLUSIVE. At prod scale, do it by hand with CONCURRENTLY instead.
+
+CREATE INDEX IF NOT EXISTS sso_expiry_ri_ix
+    ON public.sso_expiry ((network_id::text), (device_id::text));
+
+CREATE INDEX IF NOT EXISTS sso_expiry_lookup_ix
+    ON public.sso_expiry (network_id, device_id, nonce_expiration);
+
+CREATE INDEX IF NOT EXISTS sso_expiry_creation_time_ix
+    ON public.sso_expiry (creation_time);
+
+DROP INDEX IF EXISTS sso_expiry_network_member_ix;

node/Metrics.cpp

@@ -144,6 +144,8 @@ prometheus::simpleapi::counter_metric_t pgsql_mem_notification { "controller_pgs
 prometheus::simpleapi::counter_metric_t pgsql_net_notification { "controller_pgsql_network_notifications_received", "number of network change notifications received via pgsql NOTIFY" };
 prometheus::simpleapi::counter_metric_t pgsql_node_checkin { "controller_pgsql_node_checkin_count", "number of node check-ins (pgsql)" };
 prometheus::simpleapi::counter_metric_t pgsql_commit_ticks { "controller_pgsql_commit_ticks", "number of commit ticks run (pgsql)" };
+prometheus::simpleapi::gauge_metric_t db_commit_queue_size { "controller_db_commit_queue", "number of entries in the controller db commit queue" };
+prometheus::simpleapi::gauge_metric_t db_commit_latency_ms { "controller_db_commit_latency_ms", "milliseconds between enqueue and commit for the most recently dequeued db change" };
 prometheus::simpleapi::counter_metric_t db_get_sso_info { "controller_db_get_sso_info", "counter" };
 
 prometheus::simpleapi::counter_metric_t redis_mem_notification { "controller_redis_member_notifications_received", "number of member change notifications received via redis" };

node/Metrics.hpp

@@ -142,6 +142,8 @@ extern prometheus::simpleapi::counter_metric_t pgsql_mem_notification;
 extern prometheus::simpleapi::counter_metric_t pgsql_net_notification;
 extern prometheus::simpleapi::counter_metric_t pgsql_node_checkin;
 extern prometheus::simpleapi::counter_metric_t pgsql_commit_ticks;
+extern prometheus::simpleapi::gauge_metric_t db_commit_queue_size;
+extern prometheus::simpleapi::gauge_metric_t db_commit_latency_ms;
 extern prometheus::simpleapi::counter_metric_t db_get_sso_info;
 
 extern prometheus::simpleapi::counter_metric_t redis_mem_notification;

nonfree/controller/BigTableStatusWriter.cpp

@@ -9,12 +9,18 @@
 #include "CtlUtil.hpp"
 #include "PubSubWriter.hpp"
 
+#include <algorithm>
+#include <atomic>
 #include <chrono>
+#include <cstddef>
 #include <functional>
 #include <google/cloud/bigtable/mutations.h>
 #include <google/cloud/bigtable/row.h>
 #include <google/cloud/bigtable/table.h>
+#include <google/cloud/status.h>
+#include <iterator>
 #include <opentelemetry/trace/provider.h>
+#include <thread>
 
 namespace cbt = google::cloud::bigtable;
 
@@ -38,6 +44,16 @@ static const int64_t kNodeInfoRefreshMs = 6LL * 24 * 60 * 60 * 1000;	// 6 days
 // offline), so the cache tracks the active set rather than every node ever seen.
 static const int64_t kCacheEntryTtlMs = 24LL * 60 * 60 * 1000;	// 24 hours
 static const int64_t kEvictionIntervalMs = 60LL * 60 * 1000;	// sweep at most hourly
+// Bigtable's MutateRows API rejects requests containing more than 100,000
+// mutations (SetCells) total, and google-cloud-cpp sends a BulkMutation as a
+// single request -- it never splits by mutation count.  Send small batches
+// (500 rows x <=5 SetCells each, far below the limit) from a small pool of
+// worker threads.  Each worker uses the *synchronous* BulkApply on purpose:
+// the async client path makes progress only on the connection's shared
+// background event threads, which stall for tens of seconds at a time in this
+// environment, while a sync call is driven entirely by its own thread.
+static const size_t kRowsPerBatch = 500;
+static const size_t kWriterThreads = 8;
 
 BigTableStatusWriter::BigTableStatusWriter(
 	const std::string& project_id,
@@ -110,88 +126,174 @@ void BigTableStatusWriter::writePending()
 	const int64_t nowMs = OSUtils::now();
 	const std::hash<std::string> hasher;
 
-	cbt::BulkMutation bulk;
-	// Row-key hash for each mutation, aligned with bulk's indices, so a failed
-	// mutation can invalidate its cache entry (FailedMutation::original_index()).
-	std::vector<uint64_t> bulkRowHashes;
-	bulkRowHashes.reserve(toWrite.size());
+	// Build every batch on this thread first (all _lastNodeInfo reads and
+	// updates stay here, lock-free), then fan the RPCs out to worker threads.
+	// Workers only perform the RPC and record its outcome; the results --
+	// cache invalidation, re-queueing, logging -- are processed back on this
+	// thread after the workers are joined.
+	struct WriteBatch {
+		cbt::BulkMutation bulk;
+		// Row-key hash per row, aligned with the batch's indices, so a failed
+		// mutation can invalidate its cache entry
+		// (FailedMutation::original_index() is relative to this batch's request).
+		std::vector<uint64_t> rowHashes;
+		size_t entryStart = 0;	 // index into toWrite of this batch's first entry
+		size_t entryCount = 0;
+		size_t mutationCount = 0;
+		std::vector<cbt::FailedMutation> failures;
+		bool threw = false;
+		std::string error;
+		int64_t durationMs = 0;
+	};
+	std::vector<WriteBatch> batches;
+	batches.reserve((toWrite.size() + kRowsPerBatch - 1) / kRowsPerBatch);
 
-	for (const auto& entry : toWrite) {
-		std::string row_key = entry.network_id + "#" + entry.node_id;
-		const uint64_t keyHash = hasher(row_key);
+	for (size_t batchStart = 0; batchStart < toWrite.size(); batchStart += kRowsPerBatch) {
+		const size_t batchEnd = std::min(batchStart + kRowsPerBatch, toWrite.size());
 
-		// Use the member's last-seen time (epoch ms) as the explicit cell timestamp for
-		// every cell in this row.  With a server-set timestamp each SetCell lands at a
-		// fresh server clock value, so re-applying the same mutation creates a *new* cell
-		// version instead of overwriting -- non-idempotent, and google-cloud-cpp won't
-		// retry such writes.  A deterministic timestamp makes a re-applied update an exact
-		// overwrite (idempotent + safely retriable).  Bigtable cell timestamps are
-		// millisecond-granular, so last_seen (already ms) aligns exactly; it's also
-		// monotonic per member, so a later check-in always reads as the newest version.
-		const std::chrono::milliseconds cellTs(entry.last_seen);
+		WriteBatch batch;
+		batch.entryStart = batchStart;
+		batch.entryCount = batchEnd - batchStart;
+		batch.rowHashes.reserve(batch.entryCount);
 
-		cbt::SingleRowMutation m(row_key);
+		for (size_t i = batchStart; i < batchEnd; ++i) {
+			const auto& entry = toWrite[i];
 
-		// node_info (os/arch/version) changes rarely.  Write it only when our
-		// last-written value for this row differs, or hasn't been refreshed in a
-		// while -- no read RPC, the controller is the sole writer of node_info.
-		const uint64_t valueHash = hasher(entry.os + "|" + entry.arch + "|" + entry.version);
-		auto it = _lastNodeInfo.find(keyHash);
-		const bool writeNodeInfo = (it == _lastNodeInfo.end()) || (it->second.valueHash != valueHash)
-			|| ((nowMs - it->second.lastWrittenMs) > kNodeInfoRefreshMs);
+			std::string row_key = entry.network_id + "#" + entry.node_id;
+			const uint64_t keyHash = hasher(row_key);
 
-		if (writeNodeInfo) {
-			m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, osColumn, cellTs, entry.os));
-			m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, archColumn, cellTs, entry.arch));
-			m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, versionColumn, cellTs, entry.version));
-			_lastNodeInfo[keyHash] = NodeInfoState { valueHash, nowMs, nowMs };
-		}
-		else {
-			it->second.lastSeenMs = nowMs;
+			// Use the member's last-seen time (epoch ms) as the explicit cell timestamp for
+			// every cell in this row.  With a server-set timestamp each SetCell lands at a
+			// fresh server clock value, so re-applying the same mutation creates a *new* cell
+			// version instead of overwriting -- non-idempotent, and google-cloud-cpp won't
+			// retry such writes.  A deterministic timestamp makes a re-applied update an exact
+			// overwrite (idempotent + safely retriable).  Bigtable cell timestamps are
+			// millisecond-granular, so last_seen (already ms) aligns exactly; it's also
+			// monotonic per member, so a later check-in always reads as the newest version.
+			// It also makes duplicate rows racing across parallel batches order-independent.
+			const std::chrono::milliseconds cellTs(entry.last_seen);
+
+			cbt::SingleRowMutation m(row_key);
+			size_t rowMutations = 0;
+
+			// node_info (os/arch/version) changes rarely.  Write it only when our
+			// last-written value for this row differs, or hasn't been refreshed in a
+			// while -- no read RPC, the controller is the sole writer of node_info.
+			const uint64_t valueHash = hasher(entry.os + "|" + entry.arch + "|" + entry.version);
+			auto it = _lastNodeInfo.find(keyHash);
+			const bool writeNodeInfo = (it == _lastNodeInfo.end()) || (it->second.valueHash != valueHash)
+				|| ((nowMs - it->second.lastWrittenMs) > kNodeInfoRefreshMs);
+
+			if (writeNodeInfo) {
+				m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, osColumn, cellTs, entry.os));
+				m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, archColumn, cellTs, entry.arch));
+				m.emplace_back(cbt::SetCell(nodeInfoColumnFamily, versionColumn, cellTs, entry.version));
+				rowMutations += 3;
+				_lastNodeInfo[keyHash] = NodeInfoState { valueHash, nowMs, nowMs };
+			}
+			else {
+				it->second.lastSeenMs = nowMs;
+			}
+
+			// check_in (ip + last_seen) changes every cycle, so it's always written.
+			char buf[64] = { 0 };
+			std::string addressStr = entry.address.toString(buf);
+			if (entry.address.ss_family == AF_INET) {
+				m.emplace_back(cbt::SetCell(checkInColumnFamily, ipv4Column, cellTs, std::move(addressStr)));
+				rowMutations++;
+			}
+			else if (entry.address.ss_family == AF_INET6) {
+				m.emplace_back(cbt::SetCell(checkInColumnFamily, ipv6Column, cellTs, std::move(addressStr)));
+				rowMutations++;
+			}
+			m.emplace_back(cbt::SetCell(checkInColumnFamily, lastSeenColumn, cellTs, entry.last_seen));
+			rowMutations++;
+
+			batch.bulk.emplace_back(m);
+			batch.rowHashes.push_back(keyHash);
+			batch.mutationCount += rowMutations;
 		}
 
-		// check_in (ip + last_seen) changes every cycle, so it's always written.
-		char buf[64] = { 0 };
-		std::string addressStr = entry.address.toString(buf);
-		if (entry.address.ss_family == AF_INET) {
-			m.emplace_back(cbt::SetCell(checkInColumnFamily, ipv4Column, cellTs, std::move(addressStr)));
-		}
-		else if (entry.address.ss_family == AF_INET6) {
-			m.emplace_back(cbt::SetCell(checkInColumnFamily, ipv6Column, cellTs, std::move(addressStr)));
-		}
-		m.emplace_back(cbt::SetCell(checkInColumnFamily, lastSeenColumn, cellTs, entry.last_seen));
-
-		bulk.emplace_back(m);
-		bulkRowHashes.push_back(keyHash);
+		batches.push_back(std::move(batch));
 	}
 
-	ZTC_LOG("Applying %zu mutations to BigTable\n", bulk.size());
+	ZTC_LOG("Applying %zu rows to BigTable in %zu batches of up to %zu\n", toWrite.size(), batches.size(), kRowsPerBatch);
 
-	try {
-		std::vector<cbt::FailedMutation> failures = _table->BulkApply(std::move(bulk));
-		ZTC_LOG("BigTable write completed with %zu failures\n", failures.size());
-		for (auto const& r : failures) {
+	std::atomic<size_t> nextBatch { 0 };
+	std::vector<std::thread> workers;
+	workers.reserve(std::min(kWriterThreads, batches.size()));
+	for (size_t t = 0; t < std::min(kWriterThreads, batches.size()); ++t) {
+		workers.emplace_back([this, &batches, &nextBatch]() {
+			for (size_t i = nextBatch.fetch_add(1); i < batches.size(); i = nextBatch.fetch_add(1)) {
+				WriteBatch& b = batches[i];
+				const auto start = std::chrono::steady_clock::now();
+				try {
+					b.failures = _table->BulkApply(std::move(b.bulk));
+				}
+				catch (const std::exception& e) {
+					b.threw = true;
+					b.error = e.what();
+				}
+				b.durationMs =
+					std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start)
+						.count();
+			}
+		});
+	}
+	for (auto& w : workers) {
+		w.join();
+	}
+
+	// Failed entries are re-queued for the next cycle rather than dropped: the
+	// deterministic cell timestamps make replays idempotent and
+	// requeuePendingStatus caps the backlog.  This covers transient
+	// per-mutation failures (expired auth tokens, UNAVAILABLE, ...) that the
+	// client does not retry internally.
+	std::vector<PendingStatusEntry> toRequeue;
+	size_t totalFailures = 0;
+	for (WriteBatch& batch : batches) {
+		if (batch.threw) {
+			// The batch's outcome is unknown, so clear the rows it covered to force a
+			// node_info rewrite next cycle rather than trusting an optimistic update,
+			// and re-queue all of its entries.
+			ZTC_LOG("Exception writing to BigTable after %lld ms: %s\n", (long long)batch.durationMs, batch.error.c_str());
+			span->SetAttribute("error", batch.error);
+			span->SetStatus(opentelemetry::trace::StatusCode::kError, batch.error);
+			for (uint64_t keyHash : batch.rowHashes) {
+				_lastNodeInfo.erase(keyHash);
+			}
+			toRequeue.insert(
+				toRequeue.end(), std::make_move_iterator(toWrite.begin() + static_cast<std::ptrdiff_t>(batch.entryStart)),
+				std::make_move_iterator(
+					toWrite.begin() + static_cast<std::ptrdiff_t>(batch.entryStart + batch.entryCount)));
+			continue;
+		}
+		ZTC_LOG(
+			"BigTable batch of %zu rows (%zu mutations) completed with %zu failures in %lld ms\n",
+			batch.rowHashes.size(), batch.mutationCount, batch.failures.size(), (long long)batch.durationMs);
+		totalFailures += batch.failures.size();
+		for (auto const& r : batch.failures) {
 			std::cerr << ::ZeroTier::controllerLogId() << " Error writing to BigTable: " << r.status() << "\n";
+			const int idx = r.original_index();
+			if (idx < 0 || static_cast<size_t>(idx) >= batch.rowHashes.size()) {
+				continue;
+			}
 			// Drop the cache entry for any failed row so its node_info is rewritten
 			// next cycle rather than being assumed durably written.
-			const int idx = r.original_index();
-			if (idx >= 0 && static_cast<size_t>(idx) < bulkRowHashes.size()) {
-				_lastNodeInfo.erase(bulkRowHashes[idx]);
+			_lastNodeInfo.erase(batch.rowHashes[idx]);
+			// Re-queue the failed entry unless the server called it malformed --
+			// batches are far below the request limits, so INVALID_ARGUMENT here
+			// means the row itself is unprocessable and would just loop forever.
+			if (r.status().code() != google::cloud::StatusCode::kInvalidArgument) {
+				toRequeue.push_back(std::move(toWrite[batch.entryStart + static_cast<size_t>(idx)]));
 			}
 		}
 	}
-	catch (const std::exception& e) {
-		ZTC_LOG("Exception writing to BigTable: %s\n", e.what());
-		span->SetAttribute("error", e.what());
-		span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what());
-		// The batch's outcome is unknown, so clear the rows it covered to force a
-		// node_info rewrite next cycle rather than trusting an optimistic update.
-		for (uint64_t keyHash : bulkRowHashes) {
-			_lastNodeInfo.erase(keyHash);
-		}
-		// Don't drop the batch — re-queue it for the next cycle.
-		requeuePendingStatus(_pending, _lock, std::move(toWrite), "BigTableStatusWriter");
-		return;
+
+	ZTC_LOG("BigTable write completed with %zu failures\n", totalFailures);
+	if (! toRequeue.empty()) {
+		ZTC_LOG("BigTableStatusWriter: re-queueing %zu entries for retry next cycle\n", toRequeue.size());
+		requeuePendingStatus(_pending, _lock, std::move(toRequeue), "BigTableStatusWriter");
 	}
 
 	// Periodically evict rows we haven't seen lately so the cache tracks the

nonfree/controller/CentralDB.cpp

@@ -128,6 +128,13 @@ CentralDB::CentralDB(const Identity& myId,
 	ZTC_LOG("NOTICE: PostgreSQL waiting for initial data download..." ZT_EOL_S);
 	_waitNoticePrinted = true;
 
+	// Zero the watchdog stamps before ANY thread that reads them starts (the
+	// heartbeat thread and the PubSub listeners below both do) -- default-constructed
+	// atomics are indeterminate in C++17.
+	for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
+		_commitThreadLastActive[i].store(0, std::memory_order_relaxed);
+	}
+
 	initializeNetworks();
 	initializeMembers();
 
@@ -232,7 +239,7 @@ CentralDB::CentralDB(const Identity& myId,
 
 	// start background threads
 	for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
-		_commitThread[i] = std::thread(&CentralDB::commitThread, this);
+		_commitThread[i] = std::thread(&CentralDB::commitThread, this, (unsigned int)i);
 	}
 	_onlineNotificationThread = std::thread(&CentralDB::onlineNotificationThread, this);
 }
@@ -279,13 +286,28 @@ bool CentralDB::waitForReady()
 bool CentralDB::isReady()
 { return ((_ready == 2) && (_connected)); }
 
+bool CentralDB::commitPipelineHealthy()
+{
+	// Unhealthy when a commit thread has stopped making progress (a wedged thread
+	// means changes it holds -- and, given the shared notification locks, soon the
+	// whole pipeline -- will never commit) or when the queue is too deep to drain
+	// promptly. The PubSub listeners nack in either case so changes are redelivered
+	// instead of acked into a queue that may never apply them.
+	if (_commitQueue.size() >= (size_t)ZT_CENTRAL_CONTROLLER_COMMIT_QUEUE_NACK_DEPTH) {
+		return false;
+	}
+	const int64_t now = OSUtils::now();
+	for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
+		const int64_t last = _commitThreadLastActive[i].load(std::memory_order_relaxed);
+		if ((last > 0) && ((now - last) > ZT_CENTRAL_CONTROLLER_COMMIT_STALL_WARN_MS)) {
+			return false;
+		}
+	}
+	return true;
+}
+
 bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("CentralDB");
-	auto span = tracer->StartSpan("CentralDB::save");
-	auto scope = tracer->WithActiveSpan(span);
-
 	bool modified = false;
 	try {
 		if (! record.is_object()) {
@@ -294,9 +316,6 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 		}
 		const std::string objtype = OSUtils::jsonString(record["objtype"], "");
 		if (objtype == "network") {
-			auto span = tracer->StartSpan("CentralDB::save::network");
-			auto scope = tracer->WithActiveSpan(span);
-
 			const uint64_t nwid = OSUtils::jsonIntHex(record["id"], 0ULL);
 			if (nwid) {
 				nlohmann::json old;
@@ -311,15 +330,19 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 					auto propagator =
 						opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
 					propagator->Inject(carrier, current_ctx);
+					qi.enqueuedAt = OSUtils::now();
 					_commitQueue.post(qi);
 					modified = true;
 				}
 			}
+			else {
+				// Correct to skip (0 is never a valid id) but must never be silent:
+				// a malformed inbound message would otherwise vanish untraceably.
+				ZTC_LOG("WARNING: dropping network save with invalid id '%s'\n",
+						OSUtils::jsonString(record["id"], "?").c_str());
+			}
 		}
 		else if (objtype == "member") {
-			auto span = tracer->StartSpan("CentralDB::save::member");
-			auto scope = tracer->WithActiveSpan(span);
-
 			const uint64_t nwid = OSUtils::jsonIntHex(record["nwid"], 0ULL);
 			const uint64_t id = OSUtils::jsonIntHex(record["id"], 0ULL);
 			if ((id) && (nwid)) {
@@ -336,6 +359,7 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 					auto propagator =
 						opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
 					propagator->Inject(carrier, current_ctx);
+					qi.enqueuedAt = OSUtils::now();
 					_commitQueue.post(qi);
 
 					modified = true;
@@ -344,9 +368,16 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 					// fprintf(stderr, "no change\n");
 				}
 			}
+			else {
+				// Correct to skip (0 is never a valid member/network id) but must never
+				// be silent -- this is the one path where a delivered-and-acked change
+				// could otherwise vanish with no trace.
+				ZTC_LOG("WARNING: dropping member save with invalid ids: nwid='%s' id='%s'\n",
+						OSUtils::jsonString(record["nwid"], "?").c_str(), OSUtils::jsonString(record["id"], "?").c_str());
+			}
 		}
 		else {
-			ZTC_LOG("uhh waaat\n");
+			ZTC_LOG("WARNING: dropping save with unknown objtype '%s'\n", objtype.c_str());
 		}
 	}
 	catch (std::exception& e) {
@@ -360,13 +391,6 @@ bool CentralDB::save(nlohmann::json& record, bool notifyListeners)
 
 void CentralDB::eraseNetwork(const uint64_t networkId)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("CentralDB");
-	auto span = tracer->StartSpan("CentralDB::eraseNetwork");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[17];
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-
 	char tmp2[24];
 	waitForReady();
 	Utils::hex(networkId, tmp2);
@@ -379,6 +403,7 @@ void CentralDB::eraseNetwork(const uint64_t networkId)
 	auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 	auto propagator = opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
 	propagator->Inject(carrier, current_ctx);
+	qi.enqueuedAt = OSUtils::now();
 	_commitQueue.post(qi);
 	// The commit thread fires _networkChanged once with the real (cached) old config when
 	// it processes the _delete_network item, so we don't notify here — doing so would
@@ -387,14 +412,10 @@ void CentralDB::eraseNetwork(const uint64_t networkId)
 
 void CentralDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("CentralDB");
-	auto span = tracer->StartSpan("CentralDB::eraseMember");
-	auto scope = tracer->WithActiveSpan(span);
 	char networkIdStr[17];
 	char memberIdStr[11];
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-	span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
+	Utils::hex(networkId, networkIdStr);
+	Utils::hex10(memberId, memberIdStr);
 
 	waitForReady();
 
@@ -407,6 +428,7 @@ void CentralDB::eraseMember(const uint64_t networkId, const uint64_t memberId)
 	auto current_ctx = opentelemetry::context::RuntimeContext::GetCurrent();
 	auto propagator = opentelemetry::context::propagation::GlobalTextMapPropagator::GetGlobalPropagator();
 	propagator->Inject(carrier, current_ctx);
+	qi.enqueuedAt = OSUtils::now();
 	_commitQueue.post(qi);
 	// The commit thread fires _memberChanged once with the real (cached) old config when
 	// it processes the _delete_member item, so we don't notify here — doing so would
@@ -418,20 +440,9 @@ void CentralDB::nodeIsOnline(const uint64_t networkId,
 							 const InetAddress& physicalAddress,
 							 const char* osArch)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("CentralDB");
-	auto span = tracer->StartSpan("CentralDB::nodeIsOnline");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[17];
-	char memberIdStr[11];
-	char ipStr[INET6_ADDRSTRLEN];
 	// osArch may be NULL (public virtual taking a raw const char*); guard so the
-	// SetAttribute / std::string construction below can't hit UB on the hot heartbeat path.
+	// std::string construction below can't hit UB on the hot heartbeat path.
 	const char* oa = osArch ? osArch : "unknown/unknown";
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-	span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
-	span->SetAttribute("physical_address", physicalAddress.toString(ipStr));
-	span->SetAttribute("os_arch", oa);
 
 	std::lock_guard<std::mutex> l(_lastOnline_l);
 	NodeOnlineRecord& i = _lastOnline[std::pair<uint64_t, uint64_t>(networkId, memberId)];
@@ -1055,6 +1066,7 @@ void CentralDB::initializeMembers()
 
 void CentralDB::heartbeat()
 {
+	setCurrentThreadName("ctl-heartbeat");
 	char publicId[1024];
 	char hostnameTmp[1024];
 	_myId.toString(false, publicId);
@@ -1127,6 +1139,49 @@ void CentralDB::heartbeat()
 			ZTC_LOG("ERROR: Redis error in heartbeat thread: %s\n", e.what());
 		}
 
+		// Commit-pipeline liveness watchdog. A commit thread that stops making
+		// progress is wedged (historically: a lock cycle in the change-notification
+		// fan-out) and the process becomes a zombie that keeps acking PubSub
+		// messages into a queue nothing drains. Restarting is the only safe
+		// recovery, so log loudly and abort; the orchestrator brings up a fresh pod
+		// that reloads state from the DB.
+		//
+		// Abort only when EVERY started commit thread is past the threshold (the
+		// incident signature: a pipeline-wide freeze) so a single legitimately slow
+		// item -- e.g. the member-deauthorize fan-out of a huge network deletion --
+		// can't kill a controller whose other threads are still draining. A single
+		// thread holding one item hostage far past that (3x) is never legitimate
+		// and also aborts.
+		{
+			const int64_t wdNow = OSUtils::now();
+			int started = 0;
+			int overAbort = 0;
+			int64_t maxStalled = 0;
+			for (int i = 0; i < ZT_CENTRAL_CONTROLLER_COMMIT_THREADS; ++i) {
+				const int64_t last = _commitThreadLastActive[i].load(std::memory_order_relaxed);
+				if (last <= 0)
+					continue;
+				++started;
+				const int64_t stalled = wdNow - last;
+				if (stalled > maxStalled)
+					maxStalled = stalled;
+				if (stalled > ZT_CENTRAL_CONTROLLER_COMMIT_STALL_ABORT_MS)
+					++overAbort;
+				if (stalled > ZT_CENTRAL_CONTROLLER_COMMIT_STALL_WARN_MS) {
+					ZTC_LOG("WARNING: commit thread %d has made no progress in %lld ms (commit queue depth %llu)\n", i,
+							(long long)stalled, (unsigned long long)_commitQueue.size());
+				}
+			}
+			if ((started > 0)
+				&& ((overAbort == started) || (maxStalled > (3LL * ZT_CENTRAL_CONTROLLER_COMMIT_STALL_ABORT_MS)))) {
+				ZTC_LOG("FATAL: commit pipeline wedged (%d/%d threads stalled past %d ms, worst %lld ms, queue depth "
+						"%llu); aborting so the controller restarts instead of silently dropping changes\n",
+						overAbort, started, ZT_CENTRAL_CONTROLLER_COMMIT_STALL_ABORT_MS, (long long)maxStalled,
+						(unsigned long long)_commitQueue.size());
+				abort();
+			}
+		}
+
 		span->End();
 		std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 	}
@@ -1145,8 +1200,10 @@ void CentralDB::_requeueFailedCommit(_queueItem& qitem)
 		return;
 	}
 	if (qitem.retryCount >= ZT_CENTRAL_CONTROLLER_MAX_COMMIT_RETRIES) {
-		ZTC_LOG("ERROR: dropping %s change after %d failed commit attempts\n",
-				OSUtils::jsonString(qitem.jsonData["objtype"], "?").c_str(), qitem.retryCount);
+		ZTC_LOG("ERROR: dropping %s change for nwid=%s id=%s after %d failed commit attempts\n",
+				OSUtils::jsonString(qitem.jsonData["objtype"], "?").c_str(),
+				OSUtils::jsonString(qitem.jsonData["nwid"], "-").c_str(),
+				OSUtils::jsonString(qitem.jsonData["id"], "-").c_str(), qitem.retryCount);
 		return;
 	}
 	++qitem.retryCount;
@@ -1167,11 +1224,24 @@ void CentralDB::_requeueFailedCommit(_queueItem& qitem)
 	_commitQueue.post(qitem);
 }
 
-void CentralDB::commitThread()
+void CentralDB::commitThread(unsigned int idx)
 {
+	setCurrentThreadName("ctl-commit");
 	ZTC_LOG("commitThread start\n");
 	_queueItem qitem;
-	while (_commitQueue.get(qitem) && (_run == 1)) {
+	while (_run == 1) {
+		// Stamp progress on every pass -- idle timeouts included -- so the watchdog
+		// in heartbeat() can tell a wedged thread from a quiet one.
+		_commitThreadLastActive[idx].store(OSUtils::now(), std::memory_order_relaxed);
+		Metrics::db_commit_queue_size = _commitQueue.size();
+		const BlockingQueue<_queueItem>::TimedWaitResult wr = _commitQueue.get(qitem, 1000);
+		if (wr == BlockingQueue<_queueItem>::STOP)
+			break;
+		if (wr == BlockingQueue<_queueItem>::TIMED_OUT)
+			continue;
+		_commitThreadLastActive[idx].store(OSUtils::now(), std::memory_order_relaxed);
+		if (qitem.enqueuedAt > 0)
+			Metrics::db_commit_latency_ms = OSUtils::now() - qitem.enqueuedAt;
 		auto provider = opentelemetry::trace::Provider::GetTracerProvider();
 		auto tracer = provider->GetTracer("CentralDB");
 
@@ -1248,21 +1318,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) {
-							ZTC_LOG("network %s does not exist.  skipping member upsert\n", networkId.c_str());
+						// Get the network's frontend. id is the PK, so this returns at most one
+						// row: present ⇒ the network exists, empty ⇒ it doesn't (a permanent
+						// condition -- e.g. nodes still join-looping a deleted network -- so skip
+						// cleanly rather than throw; the old COUNT+GROUP BY+one_row() form threw
+						// on zero rows, turning every such item into a 100-attempt retry storm
+						// that occupied the commit threads). NULL frontend reads as "".
+						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 for %s-%s\n", networkId.c_str(),
+									networkId.c_str(), memberId.c_str());
 							w.abort();
 							_pool->unborrow(c);
 							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 "
@@ -1279,6 +1350,9 @@ void CentralDB::commitThread()
 						if (! isNewMember && change_source != "controller" && frontend != change_source) {
 							// if it is not a new member and the change source is not the controller and doesn't match
 							// the frontend, don't apply the change.
+							ZTC_LOG("WARNING: dropping member change %s-%s: change_source '%s' does not match network "
+									"frontend '%s'\n",
+									networkId.c_str(), memberId.c_str(), change_source.c_str(), frontend.c_str());
 							w.abort();
 							_pool->unborrow(c);
 							continue;
@@ -1434,6 +1508,9 @@ void CentralDB::commitThread()
 						if (! isNewNetwork && change_source != "controller" && frontend != change_source) {
 							// if it is not a new network and the change source is not the controller and doesn't match
 							// the frontend, don't apply the change.
+							ZTC_LOG("WARNING: dropping network change %s: change_source '%s' does not match network "
+									"frontend '%s'\n",
+									id.c_str(), change_source.c_str(), frontend.c_str());
 							w.abort();
 							_pool->unborrow(c);
 							continue;
@@ -1630,6 +1707,7 @@ void CentralDB::commitThread()
 
 void CentralDB::onlineNotificationThread()
 {
+	setCurrentThreadName("ctl-online");
 	waitForReady();
 	while (_run == 1) {
 		{
@@ -1791,6 +1869,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,7 +7,7 @@
 #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
@@ -15,6 +15,19 @@
 // outages (e.g. an AlloyDB maintenance restart) at the cost of in-memory buffering.
 #define ZT_CENTRAL_CONTROLLER_MAX_COMMIT_RETRIES 100
 
+// Commit-thread liveness watchdog thresholds. A commit thread that stops making
+// progress (historically: wedged on a lock cycle in the change-notification
+// fan-out) turns the controller into a zombie that keeps acking PubSub messages
+// into a queue nothing drains. heartbeat() warns past the first threshold and
+// aborts the process past the second so the orchestrator restarts it.
+#define ZT_CENTRAL_CONTROLLER_COMMIT_STALL_WARN_MS 60000
+#define ZT_CENTRAL_CONTROLLER_COMMIT_STALL_ABORT_MS 300000
+
+// Above this commit-queue depth the PubSub listeners nack instead of enqueueing:
+// the backlog is beyond prompt drainage and an acked-but-uncommitted change is
+// lost if the process dies. Redelivery retries once the queue recovers.
+#define ZT_CENTRAL_CONTROLLER_COMMIT_QUEUE_NACK_DEPTH 50000
+
 #include "../../node/Metrics.hpp"
 #include "ConnectionPool.hpp"
 #include "DB.hpp"
@@ -22,6 +35,7 @@
 #include "PostgreSQL.hpp"
 #include "StatusWriter.hpp"
 
+#include <atomic>
 #include <condition_variable>
 #include <memory>
 #include <pqxx/pqxx>
@@ -57,6 +71,7 @@ class CentralDB : public DB {
 
 	virtual bool waitForReady();
 	virtual bool isReady();
+	virtual bool commitPipelineHealthy();
 	virtual bool save(nlohmann::json& record, bool notifyListeners);
 	virtual void eraseNetwork(const uint64_t networkId);
 	virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
@@ -87,7 +102,7 @@ class CentralDB : public DB {
 	void initializeMembers();
 	void heartbeat();
 
-	void commitThread();
+	void commitThread(unsigned int idx);
 	void onlineNotificationThread();
 
 	nlohmann::json _getNetworkMember(pqxx::work& tx, const std::string networkID, const std::string memberID);
@@ -109,7 +124,7 @@ class CentralDB : public DB {
 	std::string _connString;
 
 	struct _queueItem {
-		_queueItem() : jsonData(), notifyListeners(false), traceContext(), retryCount(0)
+		_queueItem() : jsonData(), notifyListeners(false), traceContext(), retryCount(0), enqueuedAt(0)
 		{
 		}
 
@@ -121,6 +136,9 @@ class CentralDB : public DB {
 		bool notifyListeners;
 		std::map<std::string, std::string> traceContext;
 		int retryCount;
+		// First-enqueue time; retries keep the original so the commit-latency gauge
+		// reflects the full enqueue-to-commit delay.
+		int64_t enqueuedAt;
 	};
 	BlockingQueue<_queueItem> _commitQueue;
 
@@ -136,6 +154,9 @@ class CentralDB : public DB {
 	std::shared_ptr<NotificationListener> _ssoAuthListener;
 	std::shared_ptr<PubSubWriter> _ssoNonceWriter;
 	std::thread _commitThread[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS];
+	// Per-commit-thread last-progress timestamps for the liveness watchdog in
+	// heartbeat(). Stamped every loop pass (idle timeouts included).
+	std::atomic<int64_t> _commitThreadLastActive[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS];
 	std::thread _onlineNotificationThread;
 
 	std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher> _lastOnline;

nonfree/controller/CtlUtil.cpp

@@ -2,8 +2,19 @@
  * See LICENSE.txt in nonfree/
  */
 
+// The two-argument pthread_setname_np is glibc's GNU extension; the strict
+// -std=c++17 make build doesn't define _GNU_SOURCE for us (the CMake build's
+// gnu++17 does). Must precede every include.
+#if defined(__linux__) && ! defined(_GNU_SOURCE)
+#define _GNU_SOURCE
+#endif
+
 #include "CtlUtil.hpp"
 
+#ifndef _WIN32
+#include <pthread.h>
+#endif
+
 namespace ZeroTier {
 
 // Defined unconditionally (outside ZT_CONTROLLER_USE_LIBPQ): the always-compiled controller
@@ -17,6 +28,18 @@ void setControllerLogId(const std::string& id)
 const char* controllerLogId()
 { return s_controllerLogId.empty() ? "----------" : s_controllerLogId.c_str(); }
 
+void setCurrentThreadName(const char* name)
+{
+	// Linux truncates at 15 chars + NUL; keep names within that.
+#if defined(__APPLE__)
+	pthread_setname_np(name);
+#elif defined(__linux__)
+	pthread_setname_np(pthread_self(), name);
+#else
+	(void)name;
+#endif
+}
+
 }	// namespace ZeroTier
 
 #ifdef ZT_CONTROLLER_USE_LIBPQ

nonfree/controller/CtlUtil.hpp

@@ -20,6 +20,10 @@ const char* _timestr();
 void setControllerLogId(const std::string& id);
 const char* controllerLogId();
 
+// Name the calling thread so /proc wait-channel and stack dumps can identify the
+// controller's thread pools. Linux limits names to 15 characters.
+void setCurrentThreadName(const char* name);
+
 std::vector<std::string> split(std::string str, char delim);
 
 std::string url_encode(const std::string& value);

nonfree/controller/DB.cpp

@@ -19,11 +19,6 @@ namespace ZeroTier {
 
 void DB::initNetwork(nlohmann::json& network)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::initNetwork");
-	auto scope = tracer->WithActiveSpan(span);
-
 	if (! network.count("private"))
 		network["private"] = true;
 	if (! network.count("creationTime"))
@@ -82,11 +77,6 @@ void DB::initNetwork(nlohmann::json& network)
 
 void DB::initMember(nlohmann::json& member)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::initMember");
-	auto scope = tracer->WithActiveSpan(span);
-
 	if (! member.count("authorized"))
 		member["authorized"] = false;
 	if (! member.count("ssoExempt"))
@@ -104,7 +94,12 @@ void DB::initMember(nlohmann::json& member)
 	if (! member.count("noAutoAssignIps"))
 		member["noAutoAssignIps"] = false;
 	if (! member.count("revision"))
-		member["revision"] = 0ULL;
+		// Start at 1 (matching Central's ztc_member.revision DEFAULT 1) so the first
+		// save() bumps to 2 and the resulting echo passes Central's strictly-less-than
+		// revision guard against a manually created row's default of 1. With 0 here the
+		// first echo carried 1 and 1 < 1 never applied -- identity/revision for manually
+		// created members never synced back to Central.
+		member["revision"] = 1ULL;
 	if (! member.count("lastDeauthorizedTime"))
 		member["lastDeauthorizedTime"] = 0ULL;
 	if (! member.count("lastAuthorizedTime"))
@@ -132,11 +127,6 @@ void DB::initMember(nlohmann::json& member)
 
 void DB::cleanNetwork(nlohmann::json& network)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::cleanNetwork");
-	auto scope = tracer->WithActiveSpan(span);
-
 	network.erase("clock");
 	network.erase("authorizedMemberCount");
 	network.erase("activeMemberCount");
@@ -146,11 +136,6 @@ void DB::cleanNetwork(nlohmann::json& network)
 
 void DB::cleanMember(nlohmann::json& member)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::cleanMember");
-	auto scope = tracer->WithActiveSpan(span);
-
 	member.erase("clock");
 	member.erase("physicalAddr");
 	member.erase("recentLog");
@@ -158,6 +143,12 @@ void DB::cleanMember(nlohmann::json& member)
 	member.erase("lastRequestMetaData");
 	member.erase("authenticationURL");		  // computed
 	member.erase("authenticationClientID");	  // computed
+	// Provenance of the PREVIOUS write, not of this one. A record read from the
+	// cache can carry the frontend's change_source (e.g. "cv1"); a controller save
+	// that inherits it is mis-attributed downstream -- CentralDB skips the
+	// controller->Central echo for non-controller sources, silently desyncing the
+	// frontend. Saves that don't set change_source default to "controller".
+	member.erase("change_source");
 }
 
 DB::DB()
@@ -169,14 +160,6 @@ DB::~DB()
 
 bool DB::get(const uint64_t networkId, nlohmann::json& network)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::getNetwork");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[32];
-	memset(networkIdStr, 0, sizeof(networkIdStr));
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-
 	waitForReady();
 	Metrics::db_get_network++;
 	std::shared_ptr<_Network> nw;
@@ -196,17 +179,6 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network)
 
 bool DB::get(const uint64_t networkId, nlohmann::json& network, const uint64_t memberId, nlohmann::json& member)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::getNetworkAndMember");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[32];
-	memset(networkIdStr, 0, sizeof(networkIdStr));
-	char memberIdStr[32];
-	memset(memberIdStr, 0, sizeof(memberIdStr));
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-	span->SetAttribute("member_id", Utils::hex(networkId, memberIdStr));
-
 	waitForReady();
 	Metrics::db_get_network_and_member++;
 	std::shared_ptr<_Network> nw;
@@ -235,17 +207,6 @@ bool DB::get(
 	nlohmann::json& member,
 	NetworkSummaryInfo& info)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::getNetworkAndMemberAndSummary");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[32];
-	memset(networkIdStr, 0, sizeof(networkIdStr));
-	char memberIdStr[32];
-	memset(memberIdStr, 0, sizeof(memberIdStr));
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-	span->SetAttribute("member_id", Utils::hex(memberId, memberIdStr));
-
 	waitForReady();
 	Metrics::db_get_network_and_member_and_summary++;
 	std::shared_ptr<_Network> nw;
@@ -271,14 +232,6 @@ bool DB::get(
 
 bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nlohmann::json>& members)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::getNetworkAndMembers");
-	auto scope = tracer->WithActiveSpan(span);
-	char networkIdStr[32];
-	memset(networkIdStr, 0, sizeof(networkIdStr));
-	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
-
 	waitForReady();
 	Metrics::db_get_member_list++;
 	std::shared_ptr<_Network> nw;
@@ -301,11 +254,6 @@ bool DB::get(const uint64_t networkId, nlohmann::json& network, std::vector<nloh
 
 void DB::networks(std::set<uint64_t>& networks)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::networks");
-	auto scope = tracer->WithActiveSpan(span);
-
 	waitForReady();
 	Metrics::db_get_network_list++;
 	std::shared_lock<std::shared_mutex> l(_networks_l);
@@ -315,11 +263,6 @@ void DB::networks(std::set<uint64_t>& networks)
 
 void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::_memberChanged");
-	auto scope = tracer->WithActiveSpan(span);
-
 	Metrics::db_member_change++;
 	uint64_t memberId = 0;
 	uint64_t networkId = 0;
@@ -410,8 +353,8 @@ void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool
 		}
 
 		if (notifyListeners) {
-			std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
-			for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
+			const std::vector<DB::ChangeListener*> listeners(_listenersSnapshot());
+			for (auto i = listeners.begin(); i != listeners.end(); ++i) {
 				(*i)->onNetworkMemberUpdate(this, networkId, memberId, memberConfig);
 			}
 		}
@@ -455,8 +398,8 @@ void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool
 	}
 
 	if ((notifyListeners) && ((wasAuth) && (! isAuth) && (networkId) && (memberId))) {
-		std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
-		for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
+		const std::vector<DB::ChangeListener*> listeners(_listenersSnapshot());
+		for (auto i = listeners.begin(); i != listeners.end(); ++i) {
 			(*i)->onNetworkMemberDeauthorize(this, networkId, memberId);
 		}
 	}
@@ -464,14 +407,6 @@ void DB::_memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool
 
 void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners)
 {
-	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());
-	span->SetAttribute("notify_listeners", notifyListeners);
-	auto scope = tracer->WithActiveSpan(span);
-
 	Metrics::db_network_change++;
 	if (notifyListeners) {
 		if (old.is_object() && old.contains("id") && networkConfig.is_object() && networkConfig.contains("id")) {
@@ -502,8 +437,8 @@ void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, boo
 				nw->config = networkConfig;
 			}
 			if (notifyListeners) {
-				std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
-				for (auto i = _changeListeners.begin(); i != _changeListeners.end(); ++i) {
+				const std::vector<DB::ChangeListener*> listeners(_listenersSnapshot());
+				for (auto i = listeners.begin(); i != listeners.end(); ++i) {
 					(*i)->onNetworkUpdate(this, networkId, networkConfig);
 				}
 			}
@@ -518,14 +453,14 @@ void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, boo
 				nlohmann::json network;
 				std::vector<nlohmann::json> members;
 				this->get(networkId, network, members);
+				const std::vector<DB::ChangeListener*> listeners(_listenersSnapshot());
 				for (auto i = members.begin(); i != members.end(); ++i) {
 					const std::string nodeID = (*i)["id"];
 					fprintf(
 						stderr, "Deauthorizing member %s on network %s due to network deletion\n", nodeID.c_str(),
 						ids.c_str());
 					const uint64_t memberId = Utils::hexStrToU64(nodeID.c_str());
-					std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
-					for (auto j = _changeListeners.begin(); j != _changeListeners.end(); ++j) {
+					for (auto j = listeners.begin(); j != listeners.end(); ++j) {
 						fprintf(
 							stderr, "Notifying listener of deauthorization of %s on %s\n", nodeID.c_str(), ids.c_str());
 						(*j)->onNetworkMemberDeauthorize(this, networkId, memberId);
@@ -546,11 +481,6 @@ void DB::_networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, boo
 
 void DB::_fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("db");
-	auto span = tracer->StartSpan("db::_fillSummaryInfo");
-	auto scope = tracer->WithActiveSpan(span);
-
 	for (auto ab = nw->activeBridgeMembers.begin(); ab != nw->activeBridgeMembers.end(); ++ab)
 		info.activeBridges.push_back(Address(*ab));
 	std::sort(info.activeBridges.begin(), info.activeBridges.end());

nonfree/controller/DB.hpp

@@ -94,6 +94,15 @@ class DB {
 	virtual bool waitForReady() = 0;
 	virtual bool isReady() = 0;
 
+	// Whether the backend can promptly and durably apply changes handed to save().
+	// Backends with an async commit pipeline override this so message listeners can
+	// defer (nack for redelivery) instead of acknowledging changes into a stalled or
+	// drowning queue, where they would be lost if the process died or wedged.
+	virtual bool commitPipelineHealthy()
+	{
+		return true;
+	}
+
 	inline bool hasNetwork(const uint64_t networkId) const
 	{
 		std::shared_lock<std::shared_mutex> l(_networks_l);
@@ -140,6 +149,18 @@ class DB {
 		_changeListeners.push_back(listener);
 	}
 
+	// Snapshot the listener list so callbacks can be invoked without holding
+	// _changeListeners_l. Listener callbacks fan out into the controller (mirror
+	// saves, netconf re-requests) and must never run under this lock: one blocked
+	// callback would otherwise wedge every thread that needs to notify, which is
+	// how the whole commit pipeline froze in prod. Listeners are only ever added,
+	// so the snapshotted pointers stay valid.
+	inline std::vector<DB::ChangeListener*> _listenersSnapshot() const
+	{
+		std::shared_lock<std::shared_mutex> l(_changeListeners_l);
+		return _changeListeners;
+	}
+
 	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
 	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
 

nonfree/controller/DBMirrorSet.cpp

@@ -4,6 +4,7 @@
 
 #include "DBMirrorSet.hpp"
 
+#include "CtlUtil.hpp"
 #include "opentelemetry/trace/provider.h"
 
 namespace ZeroTier {
@@ -11,6 +12,7 @@ namespace ZeroTier {
 DBMirrorSet::DBMirrorSet(DB::ChangeListener* listener) : _listener(listener), _running(true), _syncCheckerThread(), _dbs(), _dbs_l()
 {
 	_syncCheckerThread = std::thread([this]() {
+		setCurrentThreadName("ctl-dbsync");
 		for (;;) {
 			for (int i = 0; i < 120; ++i) {	  // 1 minute delay between checks
 				if (! _running)
@@ -230,7 +232,7 @@ bool DBMirrorSet::save(nlohmann::json& record, bool notifyListeners)
 
 	std::vector<std::shared_ptr<DB> > dbs;
 	{
-		std::unique_lock<std::shared_mutex> l(_dbs_l);
+		std::shared_lock<std::shared_mutex> l(_dbs_l);
 		dbs = _dbs;
 	}
 	if (notifyListeners) {
@@ -258,8 +260,12 @@ void DBMirrorSet::eraseNetwork(const uint64_t networkId)
 	char networkIdStr[17];
 	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
 
-	std::unique_lock<std::shared_mutex> l(_dbs_l);
-	for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
+	std::vector<std::shared_ptr<DB> > dbs;
+	{
+		std::shared_lock<std::shared_mutex> l(_dbs_l);
+		dbs = _dbs;
+	}
+	for (auto d = dbs.begin(); d != dbs.end(); ++d) {
 		(*d)->eraseNetwork(networkId);
 	}
 }
@@ -275,8 +281,12 @@ void DBMirrorSet::eraseMember(const uint64_t networkId, const uint64_t memberId)
 	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
 	span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
 
-	std::unique_lock<std::shared_mutex> l(_dbs_l);
-	for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
+	std::vector<std::shared_ptr<DB> > dbs;
+	{
+		std::shared_lock<std::shared_mutex> l(_dbs_l);
+		dbs = _dbs;
+	}
+	for (auto d = dbs.begin(); d != dbs.end(); ++d) {
 		(*d)->eraseMember(networkId, memberId);
 	}
 }
@@ -315,9 +325,17 @@ void DBMirrorSet::onNetworkUpdate(const void* db, uint64_t networkId, const nloh
 	char networkIdStr[17];
 	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
 
+	// Snapshot the db list and call save()/the listener with _dbs_l released.
+	// This runs on the DB commit threads; holding _dbs_l (a writer lock) across
+	// the fan-out blocks every reader (netconf gets/saves) and, if the listener
+	// callback ever stalls, freezes the whole commit pipeline.
 	nlohmann::json record(network);
-	std::unique_lock<std::shared_mutex> l(_dbs_l);
-	for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
+	std::vector<std::shared_ptr<DB> > dbs;
+	{
+		std::shared_lock<std::shared_mutex> l(_dbs_l);
+		dbs = _dbs;
+	}
+	for (auto d = dbs.begin(); d != dbs.end(); ++d) {
 		if (d->get() != db) {
 			(*d)->save(record, false);
 		}
@@ -336,9 +354,14 @@ void DBMirrorSet::onNetworkMemberUpdate(const void* db, uint64_t networkId, uint
 	span->SetAttribute("network_id", Utils::hex(networkId, networkIdStr));
 	span->SetAttribute("member_id", Utils::hex10(memberId, memberIdStr));
 
+	// Same pattern as onNetworkUpdate: fan out with _dbs_l released.
 	nlohmann::json record(member);
-	std::unique_lock<std::shared_mutex> l(_dbs_l);
-	for (auto d = _dbs.begin(); d != _dbs.end(); ++d) {
+	std::vector<std::shared_ptr<DB> > dbs;
+	{
+		std::shared_lock<std::shared_mutex> l(_dbs_l);
+		dbs = _dbs;
+	}
+	for (auto d = dbs.begin(); d != dbs.end(); ++d) {
 		if (d->get() != db) {
 			(*d)->save(record, false);
 		}

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 {
 
@@ -259,11 +262,6 @@ static json _renderRule(ZT_VirtualNetworkRule& rule)
 
 static bool _parseRule(json& r, ZT_VirtualNetworkRule& rule)
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("embedded_controller");
-	auto span = tracer->StartSpan("embedded_controller::parseRule");
-	auto scope = tracer->WithActiveSpan(span);
-
 	if (! r.is_object())
 		return false;
 
@@ -740,25 +738,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)
@@ -1632,12 +1653,34 @@ void EmbeddedNetworkController::onNetworkUpdate(const void* db, uint64_t network
 	auto span = tracer->StartSpan("embedded_controller::onNetworkUpdate");
 	auto scope = tracer->WithActiveSpan(span);
 
-	// Send an update to all members of the network that are online
+	// Send an update to all members of the network that are online.
+	//
+	// Collect the targets first and re-request with _memberStatus_l released. This
+	// callback runs on the DB commit threads (via _memberChanged), and both the
+	// netconf workers and request() itself need _memberStatus_l -- holding it
+	// across a full-network fan-out stalls them, and any stall here freezes the
+	// commit pipeline behind the notification locks.
 	const int64_t now = OSUtils::now();
-	std::lock_guard<std::mutex> l(_memberStatus_l);
-	for (auto i = _memberStatus.begin(); i != _memberStatus.end(); ++i) {
-		if ((i->first.networkId == networkId) && (i->second.online(now)) && (i->second.lastRequestMetaData))
-			request(networkId, InetAddress(), 0, i->second.identity, i->second.lastRequestMetaData);
+	std::vector<_MemberStatusKey> targets;
+	{
+		std::lock_guard<std::mutex> l(_memberStatus_l);
+		for (auto i = _memberStatus.begin(); i != _memberStatus.end(); ++i) {
+			if ((i->first.networkId == networkId) && (i->second.online(now)) && (i->second.lastRequestMetaData))
+				targets.push_back(i->first);
+		}
+	}
+	for (auto k = targets.begin(); k != targets.end(); ++k) {
+		Identity id;
+		std::unique_ptr<Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> > md;
+		{
+			std::lock_guard<std::mutex> l(_memberStatus_l);
+			auto ms = _memberStatus.find(*k);
+			if ((ms == _memberStatus.end()) || (! ms->second.online(now)) || (! ms->second.lastRequestMetaData))
+				continue;
+			id = ms->second.identity;
+			md.reset(new Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY>(ms->second.lastRequestMetaData));
+		}
+		request(networkId, InetAddress(), 0, id, *md);
 	}
 }
 
@@ -1652,12 +1695,23 @@ void EmbeddedNetworkController::onNetworkMemberUpdate(
 	auto span = tracer->StartSpan("embedded_controller::onNetworkMemberUpdate");
 	auto scope = tracer->WithActiveSpan(span);
 
-	// Push update to member if online
+	// Push update to member if online. Copy what request() needs and call it with
+	// _memberStatus_l released -- this runs on the DB commit threads and must not
+	// hold member-status locks across the netconf machinery. find() instead of
+	// operator[] so members we have never heard from don't get empty entries.
 	try {
-		std::lock_guard<std::mutex> l(_memberStatus_l);
-		_MemberStatus& ms = _memberStatus[_MemberStatusKey(networkId, memberId)];
-		if ((ms.online(OSUtils::now())) && (ms.lastRequestMetaData))
-			request(networkId, InetAddress(), 0, ms.identity, ms.lastRequestMetaData);
+		Identity id;
+		std::unique_ptr<Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> > md;
+		{
+			std::lock_guard<std::mutex> l(_memberStatus_l);
+			auto ms = _memberStatus.find(_MemberStatusKey(networkId, memberId));
+			if ((ms != _memberStatus.end()) && (ms->second.online(OSUtils::now())) && (ms->second.lastRequestMetaData)) {
+				id = ms->second.identity;
+				md.reset(new Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY>(ms->second.lastRequestMetaData));
+			}
+		}
+		if (md)
+			request(networkId, InetAddress(), 0, id, *md);
 	}
 	catch (...) {
 	}
@@ -1675,15 +1729,21 @@ void EmbeddedNetworkController::onNetworkMemberDeauthorize(const void* db, uint6
 		(uint32_t)_node->prng(), networkId, 0, now, ZT_REVOCATION_FLAG_FAST_PROPAGATE, Address(memberId),
 		Revocation::CREDENTIAL_TYPE_COM);
 	rev.sign(_signingId);
+	// Collect the online members under the lock, then send revocations with it
+	// released: ncSendRevocation goes through the node core and wire I/O, and this
+	// callback runs on the DB commit threads.
+	std::vector<Address> online;
 	{
 		std::lock_guard<std::mutex> l(_memberStatus_l);
 		for (auto i = _memberStatus.begin(); i != _memberStatus.end(); ++i) {
-			if ((i->first.networkId == networkId) && (i->second.online(now))) {
-				_node->ncSendRevocation(Address(i->first.nodeId), rev);
-				// fprintf(stderr, "  Sent revocation to %.10llx\n", i->first.nodeId);
-			}
+			if ((i->first.networkId == networkId) && (i->second.online(now)))
+				online.push_back(Address(i->first.nodeId));
 		}
 	}
+	for (auto a = online.begin(); a != online.end(); ++a) {
+		_node->ncSendRevocation(*a, rev);
+		// fprintf(stderr, "  Sent revocation to %.10llx\n", a->toInt());
+	}
 }
 
 void EmbeddedNetworkController::_request(
@@ -1759,6 +1819,7 @@ void EmbeddedNetworkController::_request(
 		return;
 	}
 	const bool newMember = ((! member.is_object()) || (member.empty()));
+	bool identityLearned = false;
 	DB::initMember(member);
 	_MemberStatusKey msk(nwid, identity.address().toInt());
 #ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
@@ -1801,6 +1862,7 @@ void EmbeddedNetworkController::_request(
 			// If we do not yet know this member's identity, learn it.
 			char idtmp[1024];
 			member["identity"] = identity.toString(false, idtmp);
+			identityLearned = true;
 		}
 #ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
 		b4.stop();
@@ -1912,8 +1974,15 @@ void EmbeddedNetworkController::_request(
 					nwid, requestPacketId, identity.address(), NetworkController::NC_ERROR_AUTHENTICATION_REQUIRED,
 					authInfo.data(), authInfo.sizeBytes());
 			}
-			DB::cleanMember(member);
-			_db.save(member, true);
+			// Persist only when this request actually created state (first sighting or
+			// a just-learned identity). Re-saving the full record on every SSO-pending
+			// request writes a stale snapshot that can overwrite a fresher concurrent
+			// change (e.g. an authorize arriving via PubSub), and generates constant
+			// commit/echo churn for idle-but-unauthenticated members.
+			if (newMember || identityLearned) {
+				DB::cleanMember(member);
+				_db.save(member, true);
+			}
 #ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
 			b6.stop();
 #endif
@@ -1961,8 +2030,14 @@ void EmbeddedNetworkController::_request(
 	}
 	else {
 		// If they are not authorized, STOP!
-		DB::cleanMember(member);
-		_db.save(member, true);
+		// Persist only when this request created state (first sighting / identity just
+		// learned). The old unconditional save wrote a full stale snapshot per denied
+		// request, which could race with and overwrite a concurrent authorize arriving
+		// via PubSub -- last-writer-wins with a stale writer.
+		if (newMember || identityLearned) {
+			DB::cleanMember(member);
+			_db.save(member, true);
+		}
 		_sender->ncSendError(
 			nwid, requestPacketId, identity.address(), NetworkController::NC_ERROR_ACCESS_DENIED, nullptr, 0);
 #ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
@@ -2460,8 +2535,10 @@ void EmbeddedNetworkController::_request(
 	c10++;
 	b10.start();
 #endif
-	member["change_source"] = "controller";
+	// cleanMember first: it strips any inherited change_source (provenance of the
+	// previous write); the explicit controller tag set after must survive.
 	DB::cleanMember(member);
+	member["change_source"] = "controller";
 	_db.save(member, true);
 #ifdef CENTRAL_CONTROLLER_REQUEST_BENCHMARK
 	b10.stop();
@@ -2481,18 +2558,19 @@ void EmbeddedNetworkController::_request(
 
 void EmbeddedNetworkController::_startThreads()
 {
-	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
-	auto tracer = provider->GetTracer("embedded_network_controller");
-	auto span = tracer->StartSpan("embedded_network_controller::_startThreads");
-	auto scope = tracer->WithActiveSpan(span);
-
 	std::lock_guard<std::mutex> l(_threads_l);
 	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]() {
+			setCurrentThreadName("ctl-netconf");
 			Metrics::network_config_request_threads++;
 			uint64_t processedCount = 0;
 			uint64_t idleIterations = 0;
@@ -2507,6 +2585,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++;
@@ -2540,6 +2625,7 @@ void EmbeddedNetworkController::_startThreads()
 
 void EmbeddedNetworkController::_ssoExpiryThread()
 {
+	setCurrentThreadName("ctl-sso-expiry");
 	while (_ssoExpiryRunning) {
 		auto provider = opentelemetry::trace::Provider::GetTracerProvider();
 		auto tracer = provider->GetTracer("embedded_network_controller");

nonfree/controller/EmbeddedNetworkController.hpp

@@ -26,6 +26,7 @@
 #include <string>
 #include <thread>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 namespace ZeroTier {
@@ -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/PostgreSQL.cpp

@@ -81,6 +81,7 @@ void PostgresMemberListener::reconnect()
 
 void PostgresMemberListener::listen()
 {
+	setCurrentThreadName("ctl-pg-member");
 	while (_run) {
 		// await_notification dispatches into onNotification, which can throw on a
 		// malformed payload; libpqxx propagates that out here. An exception escaping
@@ -215,6 +216,7 @@ void PostgresNetworkListener::reconnect()
 
 void PostgresNetworkListener::listen()
 {
+	setCurrentThreadName("ctl-pg-network");
 	while (_run) {
 		// See PostgresMemberListener::listen — an exception escaping this thread would
 		// terminate the controller, so everything must be caught here; a broken

nonfree/controller/PostgreSQL.hpp

@@ -42,9 +42,43 @@ class PostgresConnection : public Connection {
 	int a;
 };
 
+// Add client-side liveness/timeout parameters to a libpq connection string unless
+// the operator already set them. Without these a connection whose peer silently
+// vanishes (AlloyDB failover, network partition) blocks its thread in a socket read
+// indefinitely; with them the call fails within a bounded time and the normal
+// error/retry paths take over. Handles both key/value conninfo and postgres:// URIs.
+inline std::string hardenPostgresConnString(std::string cs)
+{
+	const bool isUri = (cs.rfind("postgres://", 0) == 0) || (cs.rfind("postgresql://", 0) == 0);
+	auto add = [&cs, isUri](const char* key, const char* kvForm, const char* uriForm) {
+		if (cs.find(key) != std::string::npos)
+			return;
+		if (isUri) {
+			cs += (cs.find('?') == std::string::npos) ? "?" : "&";
+			cs += uriForm;
+		}
+		else {
+			cs += " ";
+			cs += kvForm;
+		}
+	};
+	add("connect_timeout", "connect_timeout=10", "connect_timeout=10");
+	add("keepalives", "keepalives=1 keepalives_idle=30 keepalives_interval=10 keepalives_count=3",
+		"keepalives=1&keepalives_idle=30&keepalives_interval=10&keepalives_count=3");
+	// Bounds transmit-side hangs (peer never ACKs) at the TCP layer; milliseconds.
+	add("tcp_user_timeout", "tcp_user_timeout=30000", "tcp_user_timeout=30000");
+	// Deliberately NO statement_timeout here: it would also cap the startup bulk
+	// loads (initializeNetworks/initializeMembers stream the full data set in one
+	// COPY, and a timeout there exits the process -> crash loop on a slow/degraded
+	// DB). Dead-peer hangs -- the incident class -- are covered by the keepalive and
+	// TCP settings above; slow-but-alive statements surface via the commit-thread
+	// watchdog instead.
+	return cs;
+}
+
 class PostgresConnFactory : public ConnectionFactory {
   public:
-	PostgresConnFactory(std::string& connString) : m_connString(connString)
+	PostgresConnFactory(std::string& connString) : m_connString(hardenPostgresConnString(connString))
 	{
 	}
 

nonfree/controller/PubSubListener.cpp

@@ -97,6 +97,7 @@ PubSubListener::~PubSubListener()
 
 void PubSubListener::subscribe()
 {
+	setCurrentThreadName("ctl-pubsub-sub");
 	while (_run) {
 		try {
 			ZTC_LOG("PubSubListener::subscribe: starting session for subscription %s\n", _subscription_id.c_str());
@@ -240,6 +241,19 @@ NotificationResult PubSubNetworkListener::onNotification(const std::string& payl
 	auto span = tracer->StartSpan("PubSubNetworkListener::onNotification");
 	auto scope = tracer->WithActiveSpan(span);
 
+	// See PubSubMemberListener::onNotification -- don't ack into an unhealthy
+	// commit pipeline. Log rate-limited.
+	if (! _db->commitPipelineHealthy()) {
+		static std::atomic<int64_t> lastLog(0);
+		const int64_t now = OSUtils::now();
+		int64_t prev = lastLog.load(std::memory_order_relaxed);
+		if (((now - prev) > 5000) && lastLog.compare_exchange_strong(prev, now)) {
+			ZTC_LOG("PubSubNetworkListener: commit pipeline stalled or over depth limit; nacking for redelivery\n");
+		}
+		span->SetStatus(opentelemetry::trace::StatusCode::kError, "commit pipeline unhealthy; redeliver");
+		return NotificationResult::TransientFailure;
+	}
+
 	pbmessages::NetworkChange nc;
 	if (! nc.ParseFromString(payload)) {
 		ZTC_LOG("Failed to parse NetworkChange protobuf message\n");
@@ -252,12 +266,12 @@ NotificationResult PubSubNetworkListener::onNotification(const std::string& payl
 		nlohmann::json oldConfig, newConfig;
 
 		if (nc.has_old()) {
-			ZTC_LOG("has old network config\n");
+			ZTC_LOG("network change: has old config for %s\n", nc.old().network_id().c_str());
 			oldConfig = toJson(nc.old(), nc.change_source());
 		}
 
 		if (nc.has_new_()) {
-			ZTC_LOG("has new network config\n");
+			ZTC_LOG("network change: has new config for %s\n", nc.new_().network_id().c_str());
 			newConfig = toJson(nc.new_(), nc.change_source());
 		}
 
@@ -336,6 +350,22 @@ NotificationResult PubSubMemberListener::onNotification(const std::string& paylo
 	auto span = tracer->StartSpan("PubSubMemberListener::onNotification");
 	auto scope = tracer->WithActiveSpan(span);
 
+	// Never ack a change into a stalled or drowning commit pipeline -- save() only
+	// enqueues, so an ack here is a promise the commit threads may not keep. Nack
+	// for redelivery instead; the message applies once the pipeline recovers (or
+	// after the liveness watchdog restarts the process). Log rate-limited: during a
+	// stall every redelivery attempt comes back through here.
+	if (! _db->commitPipelineHealthy()) {
+		static std::atomic<int64_t> lastLog(0);
+		const int64_t now = OSUtils::now();
+		int64_t prev = lastLog.load(std::memory_order_relaxed);
+		if (((now - prev) > 5000) && lastLog.compare_exchange_strong(prev, now)) {
+			ZTC_LOG("PubSubMemberListener: commit pipeline stalled or over depth limit; nacking for redelivery\n");
+		}
+		span->SetStatus(opentelemetry::trace::StatusCode::kError, "commit pipeline unhealthy; redeliver");
+		return NotificationResult::TransientFailure;
+	}
+
 	pbmessages::MemberChange mc;
 	if (! mc.ParseFromString(payload)) {
 		ZTC_LOG("Failed to parse MemberChange protobuf message\n");
@@ -349,12 +379,14 @@ NotificationResult PubSubMemberListener::onNotification(const std::string& paylo
 		nlohmann::json oldConfig, newConfig;
 
 		if (mc.has_old()) {
-			ZTC_LOG("has old member config\n");
+			ZTC_LOG("member change: has old config for %s-%s\n", mc.old().network_id().c_str(),
+					mc.old().device_id().c_str());
 			oldConfig = toJson(mc.old(), mc.change_source());
 		}
 
 		if (mc.has_new_()) {
-			ZTC_LOG("has new member config\n");
+			ZTC_LOG("member change: has new config for %s-%s\n", mc.new_().network_id().c_str(),
+					mc.new_().device_id().c_str());
 			newConfig = toJson(mc.new_(), mc.change_source());
 		}
 

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