Merge pull request #2607 from zerotier/dependabot-fixes

Fix Dependabot alerts: remove unused testcontainers, bump serde_with

ext/central-controller-docker/main.sh

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

nonfree/controller/BigTableStatusWriter.cpp

@@ -9,18 +9,12 @@
 #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;
 
@@ -44,16 +38,6 @@ 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,
@@ -126,174 +110,88 @@ void BigTableStatusWriter::writePending()
 	const int64_t nowMs = OSUtils::now();
 	const std::hash<std::string> hasher;
 
-	// 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);
+	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());
 
-	for (size_t batchStart = 0; batchStart < toWrite.size(); batchStart += kRowsPerBatch) {
-		const size_t batchEnd = std::min(batchStart + kRowsPerBatch, toWrite.size());
+	for (const auto& entry : toWrite) {
+		std::string row_key = entry.network_id + "#" + entry.node_id;
+		const uint64_t keyHash = hasher(row_key);
 
-		WriteBatch batch;
-		batch.entryStart = batchStart;
-		batch.entryCount = batchEnd - batchStart;
-		batch.rowHashes.reserve(batch.entryCount);
+		// 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);
 
-		for (size_t i = batchStart; i < batchEnd; ++i) {
-			const auto& entry = toWrite[i];
+		cbt::SingleRowMutation m(row_key);
 
-			std::string row_key = entry.network_id + "#" + entry.node_id;
-			const uint64_t keyHash = hasher(row_key);
+		// 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);
 
-			// 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;
+		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;
 		}
 
-		batches.push_back(std::move(batch));
-	}
-
-	ZTC_LOG("Applying %zu rows to BigTable in %zu batches of up to %zu\n", toWrite.size(), batches.size(), kRowsPerBatch);
-
-	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;
+		// 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)));
 		}
-		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) {
+		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);
+	}
+
+	ZTC_LOG("Applying %zu mutations to BigTable\n", bulk.size());
+
+	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::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.
-			_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)]));
+			const int idx = r.original_index();
+			if (idx >= 0 && static_cast<size_t>(idx) < bulkRowHashes.size()) {
+				_lastNodeInfo.erase(bulkRowHashes[idx]);
 			}
 		}
 	}
-
-	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");
+	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;
 	}
 
 	// Periodically evict rows we haven't seen lately so the cache tracks the

nonfree/controller/CentralDB.cpp

@@ -1791,10 +1791,6 @@ 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 8
+#define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4
 
 // 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

nonfree/controller/EmbeddedNetworkController.cpp

@@ -51,10 +51,7 @@ 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.
-// 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
+#define ZT_CONTROLLER_MAX_QUEUED_REREQUESTS 16384
 
 namespace ZeroTier {
 
@@ -743,48 +740,25 @@ void EmbeddedNetworkController::request(
 		}
 		ms.lastRequestTime = now;
 	}
-
-	// 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
+	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;
 		}
 	}
 
-	// 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);
-	}
+	_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);
 }
 
 std::string EmbeddedNetworkController::networkUpdateFromPostData(uint64_t networkID, const std::string& body)
@@ -2516,12 +2490,7 @@ void EmbeddedNetworkController::_startThreads()
 	if (! _threads.empty()) {
 		return;
 	}
-#ifdef ZT1_CENTRAL_CONTROLLER
-    // Ensure central controllers have at least 8 threads running to process network config requests
-    const long hwc = std::max((long)std::thread::hardware_concurrency(), (long)8);
-#else
 	const long hwc = std::max((long)std::thread::hardware_concurrency(), (long)1);
-#endif
 	for (long t = 0; t < hwc; ++t) {
 		_threads.emplace_back([this, t]() {
 			Metrics::network_config_request_threads++;
@@ -2538,13 +2507,6 @@ void EmbeddedNetworkController::_startThreads()
 				else if (timedWaitResult == BlockingQueue<_RQEntry*>::OK) {
 					idleIterations = 0;
 					if (qe) {
-						// Clear the de-dup mark now that this entry is dequeued, so a fresh request
-						// for the same member (e.g. a config change arriving while we process it) can
-						// queue again instead of being deduped away.
-						{
-							std::lock_guard<std::mutex> l(_pendingRequests_l);
-							_pendingRequests.erase(_MemberStatusKey(qe->nwid, qe->identity.address().toInt()));
-						}
 						try {
 							_request(qe->nwid, qe->fromAddr, qe->requestPacketId, qe->identity, qe->metaData);
 							processedCount++;

nonfree/controller/EmbeddedNetworkController.hpp

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

@@ -29,19 +29,6 @@ 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)
@@ -241,37 +228,37 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 
 	pbmessages::NetworkChange_Network* n = new pbmessages::NetworkChange_Network();
 	try {
-		n->set_network_id(OSUtils::jsonString(jfield(j, "id"), ""));
-		n->set_name(OSUtils::jsonString(jfield(j, "name"), ""));
+		n->set_network_id(OSUtils::jsonString(j["id"], ""));
+		n->set_name(OSUtils::jsonString(j["name"], ""));
 		n->set_capabilities(OSUtils::jsonDump(j.value("capabilities", "[]"), -1));
-		n->set_creation_time(OSUtils::jsonInt(jfield(j, "creationTime"), 0));
-		n->set_enable_broadcast(OSUtils::jsonBool(jfield(j, "enableBroadcast"), false));
+		n->set_creation_time(OSUtils::jsonInt(j["creationTime"], 0));
+		n->set_enable_broadcast(OSUtils::jsonBool(j["enableBroadcast"], false));
 
-		for (const auto& p : jfield(j, "ipAssignmentPools")) {
+		for (const auto& p : j["ipAssignmentPools"]) {
 			if (p.is_object()) {
 				auto pool = n->add_assignment_pools();
-				pool->set_start_ip(OSUtils::jsonString(jfield(p, "ipRangeStart"), ""));
-				pool->set_end_ip(OSUtils::jsonString(jfield(p, "ipRangeEnd"), ""));
+				pool->set_start_ip(OSUtils::jsonString(p["ipRangeStart"], ""));
+				pool->set_end_ip(OSUtils::jsonString(p["ipRangeEnd"], ""));
 			}
 		}
 
-		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));
+		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));
 
-		for (const auto& p : jfield(j, "routes")) {
+		for (const auto& p : j["routes"]) {
 			if (p.is_object()) {
 				auto r = n->add_routes();
-				r->set_target(OSUtils::jsonString(jfield(p, "target"), ""));
-				r->set_via(OSUtils::jsonString(jfield(p, "via"), ""));
+				r->set_target(OSUtils::jsonString(p["target"], ""));
+				r->set_via(OSUtils::jsonString(p["via"], ""));
 			}
 		}
 		std::string rules;
-		if (jfield(j, "rules").is_array()) {
-			rules = OSUtils::jsonDump(jfield(j, "rules"), -1);
+		if (j["rules"].is_array()) {
+			rules = OSUtils::jsonDump(j["rules"], -1);
 		}
 		else {
 			rules = "[]";
@@ -279,8 +266,8 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 		n->set_rules(rules);
 
 		std::string tags;
-		if (jfield(j, "tags").is_array()) {
-			tags = OSUtils::jsonDump(jfield(j, "tags"), -1);
+		if (j["tags"].is_array()) {
+			tags = OSUtils::jsonDump(j["tags"], -1);
 		}
 		else {
 			tags = "[]";
@@ -288,22 +275,22 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 		n->set_tags(tags);
 
 		pbmessages::NetworkChange_IPV4AssignMode* v4am = new pbmessages::NetworkChange_IPV4AssignMode();
-		if (jfield(j, "v4AssignMode").is_object()) {
-			nlohmann::json am = jfield(j, "v4AssignMode");
+		if (j["v4AssignMode"].is_object()) {
+			nlohmann::json am = 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 (jfield(j, "v6AssignMode").is_object()) {
-			nlohmann::json am = jfield(j, "v6AssignMode");
+		if (j["v6AssignMode"].is_object()) {
+			nlohmann::json am = 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 = jfield(j, "dns");
+		nlohmann::json jdns = j["dns"];
 		if (jdns.is_object()) {
 			pbmessages::NetworkChange_DNS* dns = new pbmessages::NetworkChange_DNS();
 			dns->set_domain(jdns.value("domain", ""));
@@ -316,14 +303,14 @@ pbmessages::NetworkChange_Network* networkFromJson(const nlohmann::json& j)
 			n->set_allocated_dns(dns);
 		}
 
-		n->set_sso_enabled(OSUtils::jsonBool(jfield(j, "ssoEnabled"), false));
-		nlohmann::json ssocfg = jfield(j, "ssoConfig");
+		n->set_sso_enabled(OSUtils::jsonBool(j["ssoEnabled"], false));
+		nlohmann::json ssocfg = 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(jfield(j, "rulesSource"), ""));
+		n->set_rules_source(OSUtils::jsonString(j["rulesSource"], ""));
 	}
 	catch (const std::exception& e) {
 		ZTC_LOG("Exception parsing network JSON: %s\n", e.what());
@@ -359,12 +346,12 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 
 	pbmessages::MemberChange_Member* m = new pbmessages::MemberChange_Member();
 	try {
-		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")) {
+		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"]) {
 				if (addr.is_string()) {
 					auto a = m->add_ip_assignments();
 					std::string address = addr.get<std::string>();
@@ -372,9 +359,9 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 				}
 			}
 		}
-		m->set_active_bridge(OSUtils::jsonBool(jfield(j, "activeBridge"), false));
-		if (jfield(j, "tags").is_array()) {
-			nlohmann::json tags = jfield(j, "tags");
+		m->set_active_bridge(OSUtils::jsonBool(j["activeBridge"], false));
+		if (j["tags"].is_array()) {
+			nlohmann::json tags = j["tags"];
 			std::string tagsStr = OSUtils::jsonDump(tags, -1);
 			m->set_tags(tagsStr);
 		}
@@ -383,8 +370,8 @@ pbmessages::MemberChange_Member* memberFromJson(const nlohmann::json& j)
 			std::string tagsStr = OSUtils::jsonDump(tags, -1);
 			m->set_tags(tagsStr);
 		}
-		if (jfield(j, "capabilities").is_array()) {
-			nlohmann::json caps = jfield(j, "capabilities");
+		if (j["capabilities"].is_array()) {
+			nlohmann::json caps = j["capabilities"];
 			std::string capsStr = OSUtils::jsonDump(caps, -1);
 			m->set_capabilities(capsStr);
 		}
@@ -393,21 +380,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(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));
+		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));
 	}
 	catch (const std::exception& e) {
 		ZTC_LOG("Exception parsing member JSON: %s\n", e.what());

rustybits/Cargo.toml

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

scripts/bootstrap-deps.sh

@@ -79,7 +79,6 @@ else
             -DWITH_BENCHMARK=OFF -DWITH_EXAMPLES=OFF -DWITH_FUNC_TESTS=OFF \
             -DUSE_THIRDPARTY_LIBRARIES=OFF \
             -DWITH_ABSEIL=ON \
-            -DWITH_API_ONLY=OFF \
             -DWITH_OTLP_GRPC=ON -DWITH_OTLP_HTTP=ON -DWITH_PROMETHEUS=OFF
     else
         # Header-only API only (default): quick, no SDK/exporters, no abseil/protobuf/grpc.