text: #include "zeek/broker/Manager.h"
  
  #include <broker/broker.hh>
  #include <broker/zeek.hh>
  #include <unistd.h>
  #include <cstdio>
  #include <cstring>
  
  #include "zeek/DebugLogger.h"
  #include "zeek/Desc.h"
  #include "zeek/Func.h"
  #include "zeek/IntrusivePtr.h"
  #include "zeek/Reporter.h"
  #include "zeek/RunState.h"
  #include "zeek/SerializationFormat.h"
  #include "zeek/Var.h"
  #include "zeek/broker/Data.h"
  #include "zeek/broker/Store.h"
  #include "zeek/broker/comm.bif.h"
  #include "zeek/broker/data.bif.h"
  #include "zeek/broker/messaging.bif.h"
  #include "zeek/broker/store.bif.h"
  #include "zeek/iosource/Manager.h"
  #include "zeek/logging/Manager.h"
  #include "zeek/telemetry/Manager.h"
  #include "zeek/util.h"
  
  using namespace std;
  
  namespace zeek::Broker
  	{
  
  static inline Val* get_option(const char* option)
  	{
  	const auto& id = zeek::detail::global_scope()->Find(option);
  
  	if ( ! (id && id->GetVal()) )
  		reporter->FatalError("Unknown Broker option %s", option);
  
  	return id->GetVal().get();
  	}
  
  template <class T> static inline void set_option(const char* option, const T& value)
  	{
  	const auto& id = zeek::detail::global_scope()->Find(option);
  
  	if ( ! id )
  		reporter->FatalError("Unknown Broker option %s", option);
  
  	if constexpr ( std::is_same_v<T, broker::port> )
  		{
  		switch ( value.type() )
  			{
  			case broker::port::protocol::tcp:
  				id->SetVal(val_mgr->Port(value.number(), TRANSPORT_TCP));
  				break;
  			case broker::port::protocol::udp:
  				id->SetVal(val_mgr->Port(value.number(), TRANSPORT_UDP));
  				break;
  			case broker::port::protocol::icmp:
  				id->SetVal(val_mgr->Port(value.number(), TRANSPORT_ICMP));
  				break;
  			default:
  				id->SetVal(val_mgr->Port(value.number(), TRANSPORT_UNKNOWN));
  			}
  		}
  	else if constexpr ( std::is_same_v<T, broker::timespan> )
  		{
  		using std::chrono::duration_cast;
  		auto ts = duration_cast<broker::fractional_seconds>(value);
  		id->SetVal(make_intrusive<IntervalVal>(ts.count()));
  		}
  	else if constexpr ( std::is_same_v<T, std::vector<std::string>> )
  		{
  		auto ptr = make_intrusive<VectorVal>(zeek::id::string_vec);
  		for ( const auto& str : value )
  			ptr->Append(make_intrusive<StringVal>(str));
  		id->SetVal(std::move(ptr));
  		}
  	else
  		{
  		static_assert(std::is_same_v<T, std::string>);
  		id->SetVal(make_intrusive<StringVal>(value));
  		}
  	}
  
  namespace
  	{
  
  struct opt_mapping
  	{
  	broker::configuration* cfg;
  	std::string_view broker_name;
  	const char* zeek_name;
  
  	template <class T> auto broker_read() { return caf::get_as<T>(*cfg, broker_name); }
  
  	template <class T> auto broker_write(T&& val) { cfg->set(broker_name, std::forward<T>(val)); }
  
  	auto zeek_read() { return get_option(zeek_name); }
  
  	template <class T> auto zeek_write(const T& val) { set_option(zeek_name, val); }
  	};
  
  #define WITH_OPT_MAPPING(broker_name, zeek_name)                                                   \
  	if ( auto opt = opt_mapping{&config, broker_name, zeek_name}; true )
  
  	} // namespace
  
  class BrokerConfig : public broker::configuration
  	{
  public:
  	BrokerConfig(broker::broker_options options) : broker::configuration(options)
  		{
  		openssl_cafile = get_option("Broker::ssl_cafile")->AsString()->CheckString();
  		openssl_capath = get_option("Broker::ssl_capath")->AsString()->CheckString();
  		openssl_certificate = get_option("Broker::ssl_certificate")->AsString()->CheckString();
  		openssl_key = get_option("Broker::ssl_keyfile")->AsString()->CheckString();
  		openssl_passphrase = get_option("Broker::ssl_passphrase")->AsString()->CheckString();
  		}
  	};
  
  class BrokerState
  	{
  public:
  	BrokerState(BrokerConfig config, size_t congestion_queue_size)
  		: endpoint(std::move(config)),
  		  subscriber(endpoint.make_subscriber({broker::topic::statuses(), broker::topic::errors()},
  	                                          congestion_queue_size))
  		{
  		}
  
  	broker::endpoint endpoint;
  	broker::subscriber subscriber;
  	};
  
  const broker::endpoint_info Manager::NoPeer{{}, {}};
  
  int Manager::script_scope = 0;
  
  struct scoped_reporter_location
  	{
  	scoped_reporter_location(zeek::detail::Frame* frame)
  		{
  		reporter->PushLocation(frame->GetCallLocation());
  		}
  
  	~scoped_reporter_location() { reporter->PopLocation(); }
  	};
  
  #ifdef DEBUG
  static std::string RenderMessage(std::string topic, const broker::data& x)
  	{
  	return util::fmt("%s -> %s", broker::to_string(x).c_str(), topic.c_str());
  	}
  
  static std::string RenderEvent(std::string topic, std::string name, const broker::data& args)
  	{
  	return util::fmt("%s(%s) -> %s", name.c_str(), broker::to_string(args).c_str(), topic.c_str());
  	}
  
  static std::string RenderMessage(const broker::store::response& x)
  	{
  	return util::fmt("%s [id %" PRIu64 "]",
  	                 (x.answer ? broker::to_string(*x.answer).c_str() : "<no answer>"), x.id);
  	}
  
  static std::string RenderMessage(const broker::vector* xs)
  	{
  	return broker::to_string(*xs);
  	}
  
  static std::string RenderMessage(const broker::data& d)
  	{
  	return broker::to_string(d);
  	}
  
  static std::string RenderMessage(const broker::vector& xs)
  	{
  	return broker::to_string(xs);
  	}
  
  static std::string RenderMessage(broker::status_view s)
  	{
  	return broker::to_string(s.code());
  	}
  
  static std::string RenderMessage(broker::error_view e)
  	{
  	if ( auto ctx = e.context() )
  		return util::fmt("%s (%s)", to_string(e.code()).c_str(), to_string(*ctx).c_str());
  	else
  		return util::fmt("%s (null)", to_string(e.code()).c_str());
  	}
  
  #endif
  
  Manager::Manager(bool arg_use_real_time)
  	{
  	bound_port = 0;
  	use_real_time = arg_use_real_time;
  	peer_count = 0;
  	log_batch_size = 0;
  	log_topic_func = nullptr;
  	log_id_type = nullptr;
  	writer_id_type = nullptr;
  	}
  
  Manager::~Manager() { }
  
  void Manager::InitPostScript()
  	{
  	DBG_LOG(DBG_BROKER, "Initializing");
  
  	log_batch_size = get_option("Broker::log_batch_size")->AsCount();
  	default_log_topic_prefix =
  		get_option("Broker::default_log_topic_prefix")->AsString()->CheckString();
  	log_topic_func = get_option("Broker::log_topic")->AsFunc();
  	log_id_type = id::find_type("Log::ID")->AsEnumType();
  	writer_id_type = id::find_type("Log::Writer")->AsEnumType();
  	zeek_table_manager = get_option("Broker::table_store_master")->AsBool();
  	zeek_table_db_directory =
  		get_option("Broker::table_store_db_directory")->AsString()->CheckString();
  
  	detail::opaque_of_data_type = make_intrusive<OpaqueType>("Broker::Data");
  	detail::opaque_of_set_iterator = make_intrusive<OpaqueType>("Broker::SetIterator");
  	detail::opaque_of_table_iterator = make_intrusive<OpaqueType>("Broker::TableIterator");
  	detail::opaque_of_vector_iterator = make_intrusive<OpaqueType>("Broker::VectorIterator");
  	detail::opaque_of_record_iterator = make_intrusive<OpaqueType>("Broker::RecordIterator");
  	detail::opaque_of_store_handle = make_intrusive<OpaqueType>("Broker::Store");
  	vector_of_data_type = make_intrusive<VectorType>(id::find_type("Broker::Data"));
  
  	// Register as a "dont-count" source first, we may change that later.
  	iosource_mgr->Register(this, true);
  
  	broker::broker_options options;
  	options.disable_ssl = get_option("Broker::disable_ssl")->AsBool();
  	options.forward = get_option("Broker::forward_messages")->AsBool();
  	options.use_real_time = use_real_time;
  
  	BrokerConfig config{std::move(options)};
  
  	auto scheduler_policy = get_option("Broker::scheduler_policy")->AsString()->CheckString();
  
  	if ( util::streq(scheduler_policy, "sharing") )
  		config.set("caf.scheduler.policy", "sharing");
  	else if ( util::streq(scheduler_policy, "stealing") )
  		config.set("caf.scheduler.policy", "stealing");
  	else
  		reporter->FatalError("Invalid Broker::scheduler_policy: %s", scheduler_policy);
  
  	auto max_threads_env = getenv("ZEEK_BROKER_MAX_THREADS");
  
  	if ( max_threads_env )
  		config.set("caf.scheduler.max-threads", atoi(max_threads_env));
  	else
  		config.set("caf.scheduler.max-threads", get_option("Broker::max_threads")->AsCount());
  
  	config.set("caf.work-stealing.moderate-sleep-duration",
  	           caf::timespan(static_cast<unsigned>(
  				   get_option("Broker::moderate_sleep")->AsInterval() * 1e9)));
  
  	config.set("caf.work-stealing.relaxed-sleep-duration",
  	           caf::timespan(
  				   static_cast<unsigned>(get_option("Broker::relaxed_sleep")->AsInterval() * 1e9)));
  
  	config.set("caf.work-stealing.aggressive-poll-attempts",
  	           get_option("Broker::aggressive_polls")->AsCount());
  	config.set("caf.work-stealing.moderate-poll-attempts",
  	           get_option("Broker::moderate_polls")->AsCount());
  
  	config.set("caf.work-stealing.aggressive-steal-interval",
  	           get_option("Broker::aggressive_interval")->AsCount());
  	config.set("caf.work-stealing.moderate-steal-interval",
  	           get_option("Broker::moderate_interval")->AsCount());
  	config.set("caf.work-stealing.relaxed-steal-interval",
  	           get_option("Broker::relaxed_interval")->AsCount());
  
  	// Before launching Broker, we check whether the configuration contains
  	// values for the metric_exporter_* options. If Broker already has picked up
  	// values from environment variables (or config files) then we write then
  	// back. Otherwise, we forward user-defined values from script land (but
  	// ignore defaults).
  	WITH_OPT_MAPPING("broker.metrics.port", "Broker::metrics_port")
  		{
  		if ( auto port = opt.broker_read<uint16_t>() )
  			{
  			opt.zeek_write(broker::port{*port, broker::port::protocol::tcp});
  			}
  		else
  			{
  			auto ptr = opt.zeek_read()->AsPortVal();
  			if ( ptr->IsTCP() )
  				opt.broker_write(ptr->Port());
  			}
  		}
  
  	WITH_OPT_MAPPING("broker.metrics.export.interval", "Broker::metrics_export_interval")
  		{
  		if ( auto ts = opt.broker_read<broker::timespan>() )
  			{
  			opt.zeek_write(*ts);
  			}
  		else
  			{
  			using std::chrono::duration_cast;
  			auto val = opt.zeek_read()->AsInterval();
  			auto frac_ts = broker::fractional_seconds{val};
  			if ( frac_ts.count() > 0.0 )
  				opt.broker_write(duration_cast<broker::timespan>(frac_ts));
  			}
  		}
  
  	WITH_OPT_MAPPING("broker.metrics.export.topic", "Broker::metrics_export_topic")
  		{
  		if ( auto str = opt.broker_read<std::string>() )
  			{
  			opt.zeek_write(*str);
  			}
  		else
  			{
  			auto ptr = opt.zeek_read()->AsStringVal();
  			if ( ptr->Len() > 0 )
  				opt.broker_write(ptr->ToStdString());
  			}
  		}
  
  	WITH_OPT_MAPPING("broker.metrics.endpoint-name", "Broker::metrics_export_endpoint_name")
  		{
  		if ( auto str = opt.broker_read<std::string>() )
  			{
  			opt.zeek_write(*str);
  			}
  		else
  			{
  			auto ptr = opt.zeek_read()->AsStringVal();
  			if ( ptr->Len() > 0 )
  				opt.broker_write(ptr->ToStdString());
  			}
  		}
  
  	WITH_OPT_MAPPING("broker.metrics.export.prefixes", "Broker::metrics_export_prefixes")
  		{
  		if ( auto str = opt.broker_read<std::vector<std::string>>() )
  			{
  			opt.zeek_write(*str);
  			}
  		else
  			{
  			auto ptr = opt.zeek_read()->AsVectorVal();
  			std::vector<std::string> str_ls;
  			for ( unsigned index = 0; index < ptr->Size(); ++index )
  				str_ls.emplace_back(ptr->StringValAt(index)->ToStdString());
  			opt.broker_write(std::move(str_ls));
  			}
  		}
  
  	auto cqs = get_option("Broker::congestion_queue_size")->AsCount();
  	bstate = std::make_shared<BrokerState>(std::move(config), cqs);
  
  	if ( ! iosource_mgr->RegisterFd(bstate->subscriber.fd(), this) )
  		reporter->FatalError("Failed to register broker subscriber with iosource_mgr");
  
  	bstate->subscriber.add_topic(broker::topic::store_events(), true);
  
  	InitializeBrokerStoreForwarding();
  	}
  
  void Manager::InitializeBrokerStoreForwarding()
  	{
  	const auto& globals = zeek::detail::global_scope()->Vars();
  
  	for ( const auto& global : globals )
  		{
  		auto& id = global.second;
  		if ( id->HasVal() && id->GetAttr(zeek::detail::ATTR_BACKEND) )
  			{
  			const auto& attr = id->GetAttr(zeek::detail::ATTR_BACKEND);
  			auto e = static_cast<BifEnum::Broker::BackendType>(
  				attr->GetExpr()->Eval(nullptr)->AsEnum());
  			auto storename = std::string("___sync_store_") + global.first;
  			id->GetVal()->AsTableVal()->SetBrokerStore(storename);
  			AddForwardedStore(storename, cast_intrusive<TableVal>(id->GetVal()));
  
  			// We only create masters here. For clones, we do all the work of setting up
  			// the forwarding - but we do not try to initialize the clone. We can only initialize
  			// the clone, once a node has a connection to a master. This is currently done in
  			// scriptland in scripts/base/frameworks/cluster/broker-stores.zeek. Once the ALM
  			// transport is ready we can change over to doing this here.
  			if ( ! zeek_table_manager )
  				continue;
  
  			auto backend = detail::to_backend_type(e);
  			auto suffix = ".store";
  
  			switch ( backend )
  				{
  				case broker::backend::sqlite:
  					suffix = ".sqlite";
  					break;
  				default:
  					break;
  				}
  
  			auto path = zeek_table_db_directory + "/" + storename + suffix;
  
  			MakeMaster(storename, backend, broker::backend_options{{"path", path}});
  			}
  		}
  	}
  
  void Manager::Terminate()
  	{
  	FlushLogBuffers();
  
  	iosource_mgr->UnregisterFd(bstate->subscriber.fd(), this);
  
  	vector<string> stores_to_close;
  
  	for ( auto& x : data_stores )
  		stores_to_close.push_back(x.first);
  
  	for ( auto& x : stores_to_close )
  		// This doesn't loop directly over data_stores, because CloseStore
  		// modifies the map and invalidates iterators.
  		CloseStore(x);
  
  	FlushLogBuffers();
  
  	for ( auto& p : bstate->endpoint.peers() )
  		if ( p.peer.network )
  			bstate->endpoint.unpeer(p.peer.network->address, p.peer.network->port);
  	}
  
  bool Manager::Active()
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return false;
  
  	if ( bound_port > 0 )
  		return true;
  
  	return peer_count > 0;
  	}
  
  void Manager::AdvanceTime(double seconds_since_unix_epoch)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return;
  
  	if ( bstate->endpoint.use_real_time() )
  		return;
  
  	auto secs = std::chrono::duration<double>(seconds_since_unix_epoch);
  	auto span = std::chrono::duration_cast<broker::timespan>(secs);
  	broker::timestamp next_time{span};
  	bstate->endpoint.advance_time(next_time);
  	}
  
  void Manager::FlushPendingQueries()
  	{
  	while ( ! pending_queries.empty() )
  		{
  		// possibly an infinite loop if a query can recursively
  		// generate more queries...
  		for ( auto& s : data_stores )
  			{
  			while ( ! s.second->proxy.mailbox().empty() )
  				{
  				auto response = s.second->proxy.receive();
  				ProcessStoreResponse(s.second, move(response));
  				}
  			}
  		}
  	}
  
  void Manager::ClearStores()
  	{
  	FlushPendingQueries();
  
  	for ( const auto& [name, handle] : data_stores )
  		handle->store.clear();
  	}
  
  uint16_t Manager::Listen(const string& addr, uint16_t port)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return 0;
  
  	bound_port = bstate->endpoint.listen(addr, port);
  
  	if ( bound_port == 0 )
  		Error("Failed to listen on %s:%" PRIu16, addr.empty() ? "INADDR_ANY" : addr.c_str(), port);
  
  	// Register as a "does-count" source now.
  	iosource_mgr->Register(this, false);
  
  	DBG_LOG(DBG_BROKER, "Listening on %s:%" PRIu16, addr.empty() ? "INADDR_ANY" : addr.c_str(),
  	        port);
  
  	return bound_port;
  	}
  
  void Manager::Peer(const string& addr, uint16_t port, double retry)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return;
  
  	DBG_LOG(DBG_BROKER, "Starting to peer with %s:%" PRIu16, addr.c_str(), port);
  
  	auto e = getenv("ZEEK_DEFAULT_CONNECT_RETRY");
  
  	if ( e )
  		retry = atoi(e);
  
  	if ( retry > 0.0 && retry < 1.0 )
  		// Ensure that it doesn't get turned into zero.
  		retry = 1.0;
  
  	auto secs = broker::timeout::seconds(static_cast<uint64_t>(retry));
  	bstate->endpoint.peer_nosync(addr, port, secs);
  
  	auto counts_as_iosource = get_option("Broker::peer_counts_as_iosource")->AsBool();
  
  	if ( counts_as_iosource )
  		// Register as a "does-count" source now.
  		iosource_mgr->Register(this, false);
  	}
  
  void Manager::Unpeer(const string& addr, uint16_t port)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return;
  
  	DBG_LOG(DBG_BROKER, "Stopping to peer with %s:%" PRIu16, addr.c_str(), port);
  
  	FlushLogBuffers();
  	bstate->endpoint.unpeer_nosync(addr, port);
  	}
  
  std::vector<broker::peer_info> Manager::Peers() const
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return {};
  
  	return bstate->endpoint.peers();
  	}
  
  std::string Manager::NodeID() const
  	{
  	return to_string(bstate->endpoint.node_id());
  	}
  
  bool Manager::PublishEvent(string topic, std::string name, broker::vector args)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return true;
  
  	if ( peer_count == 0 )
  		return true;
  
  	DBG_LOG(DBG_BROKER, "Publishing event: %s", RenderEvent(topic, name, args).c_str());
  	broker::zeek::Event ev(std::move(name), std::move(args));
  	bstate->endpoint.publish(move(topic), ev.move_data());
  	++statistics.num_events_outgoing;
  	return true;
  	}
  
  bool Manager::PublishEvent(string topic, RecordVal* args)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return true;
  
  	if ( peer_count == 0 )
  		return true;
  
  	if ( ! args->HasField(0) )
  		return false;
  
  	auto event_name = args->GetFieldAs<StringVal>(0)->CheckString();
  	auto vv = args->GetFieldAs<VectorVal>(1);
  	broker::vector xs;
  	xs.reserve(vv->Size());
  
  	for ( auto i = 0u; i < vv->Size(); ++i )
  		{
  		const auto& val = vv->RecordValAt(i)->GetField(0);
  		auto data_val = static_cast<detail::DataVal*>(val.get());
  		xs.emplace_back(data_val->data);
  		}
  
  	return PublishEvent(std::move(topic), event_name, std::move(xs));
  	}
  
  bool Manager::PublishIdentifier(std::string topic, std::string id)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return true;
  
  	if ( peer_count == 0 )
  		return true;
  
  	const auto& i = zeek::detail::global_scope()->Find(id);
  
  	if ( ! i )
  		return false;
  
  	const auto& val = i->GetVal();
  
  	if ( ! val )
  		// Probably could have a special case to also unset the value on the
  		// receiving side, but not sure what use that would be.
  		return false;
  
  	auto data = detail::val_to_data(val.get());
  
  	if ( ! data )
  		{
  		Error("Failed to publish ID with unsupported type: %s (%s)", id.c_str(),
  		      type_name(val->GetType()->Tag()));
  		return false;
  		}
  
  	broker::zeek::IdentifierUpdate msg(move(id), move(*data));
  	DBG_LOG(DBG_BROKER, "Publishing id-update: %s", RenderMessage(topic, msg.as_data()).c_str());
  	bstate->endpoint.publish(move(topic), msg.move_data());
  	++statistics.num_ids_outgoing;
  	return true;
  	}
  
  bool Manager::PublishLogCreate(EnumVal* stream, EnumVal* writer,
                                 const logging::WriterBackend::WriterInfo& info, int num_fields,
                                 const threading::Field* const* fields,
                                 const broker::endpoint_info& peer)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return true;
  
  	if ( peer_count == 0 )
  		return true;
  
  	auto stream_id = stream->GetType()->AsEnumType()->Lookup(stream->AsEnum());
  
  	if ( ! stream_id )
  		{
  		reporter->Error("Failed to remotely log: stream %d doesn't have name", stream->AsEnum());
  		return false;
  		}
  
  	auto writer_id = writer->GetType()->AsEnumType()->Lookup(writer->AsEnum());
  
  	if ( ! writer_id )
  		{
  		reporter->Error("Failed to remotely log: writer %d doesn't have name", writer->AsEnum());
  		return false;
  		}
  
  	auto writer_info = info.ToBroker();
  
  	broker::vector fields_data;
  	fields_data.reserve(num_fields);
  
  	for ( auto i = 0; i < num_fields; ++i )
  		{
  		auto field_data = detail::threading_field_to_data(fields[i]);
  		fields_data.push_back(move(field_data));
  		}
  
  	std::string topic = default_log_topic_prefix + stream_id;
  	auto bstream_id = broker::enum_value(move(stream_id));
  	auto bwriter_id = broker::enum_value(move(writer_id));
  	broker::zeek::LogCreate msg(move(bstream_id), move(bwriter_id), move(writer_info),
  	                            move(fields_data));
  
  	DBG_LOG(DBG_BROKER, "Publishing log creation: %s", RenderMessage(topic, msg.as_data()).c_str());
  
  	if ( peer.node != NoPeer.node )
  		// Direct message.
  		bstate->endpoint.publish(peer, move(topic), msg.move_data());
  	else
  		// Broadcast.
  		bstate->endpoint.publish(move(topic), msg.move_data());
  
  	return true;
  	}
  
  bool Manager::PublishLogWrite(EnumVal* stream, EnumVal* writer, string path, int num_fields,
                                const threading::Value* const* vals)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return true;
  
  	if ( peer_count == 0 )
  		return true;
  
  	auto stream_id_num = stream->AsEnum();
  	auto stream_id = stream->GetType()->AsEnumType()->Lookup(stream_id_num);
  
  	if ( ! stream_id )
  		{
  		reporter->Error("Failed to remotely log: stream %d doesn't have name", stream->AsEnum());
  		return false;
  		}
  
  	auto writer_id = writer->GetType()->AsEnumType()->Lookup(writer->AsEnum());
  
  	if ( ! writer_id )
  		{
  		reporter->Error("Failed to remotely log: writer %d doesn't have name", writer->AsEnum());
  		return false;
  		}
  
  	zeek::detail::BinarySerializationFormat fmt;
  	char* data;
  	int len;
  
  	fmt.StartWrite();
  
  	bool success = fmt.Write(num_fields, "num_fields");
  
  	if ( ! success )
  		{
  		reporter->Error("Failed to remotely log stream %s: num_fields serialization failed",
  		                stream_id);
  		return false;
  		}
  
  	for ( int i = 0; i < num_fields; ++i )
  		{
  		if ( ! vals[i]->Write(&fmt) )
  			{
  			reporter->Error("Failed to remotely log stream %s: field %d serialization failed",
  			                stream_id, i);
  			return false;
  			}
  		}
  
  	len = fmt.EndWrite(&data);
  	std::string serial_data(data, len);
  	free(data);
  
  	auto v = log_topic_func->Invoke(IntrusivePtr{NewRef{}, stream},
  	                                make_intrusive<StringVal>(path));
  
  	if ( ! v )
  		{
  		reporter->Error("Failed to remotely log: log_topic func did not return"
  		                " a value for stream %s at path %s",
  		                stream_id, path.data());
  		return false;
  		}
  
  	std::string topic = v->AsString()->CheckString();
  
  	auto bstream_id = broker::enum_value(move(stream_id));
  	auto bwriter_id = broker::enum_value(move(writer_id));
  	broker::zeek::LogWrite msg(move(bstream_id), move(bwriter_id), move(path), move(serial_data));
  
  	DBG_LOG(DBG_BROKER, "Buffering log record: %s", RenderMessage(topic, msg.as_data()).c_str());
  
  	if ( log_buffers.size() <= (unsigned int)stream_id_num )
  		log_buffers.resize(stream_id_num + 1);
  
  	auto& lb = log_buffers[stream_id_num];
  	++lb.message_count;
  	auto& pending_batch = lb.msgs[topic];
  	pending_batch.emplace_back(msg.move_data());
  
  	if ( lb.message_count >= log_batch_size )
  		statistics.num_logs_outgoing += lb.Flush(bstate->endpoint, log_batch_size);
  
  	return true;
  	}
  
  size_t Manager::LogBuffer::Flush(broker::endpoint& endpoint, size_t log_batch_size)
  	{
  	if ( endpoint.is_shutdown() )
  		return 0;
  
  	if ( ! message_count )
  		// No logs buffered for this stream.
  		return 0;
  
  	for ( auto& kv : msgs )
  		{
  		auto& topic = kv.first;
  		auto& pending_batch = kv.second;
  		broker::vector batch;
  		batch.reserve(log_batch_size + 1);
  		pending_batch.swap(batch);
  		broker::zeek::Batch msg(std::move(batch));
  		endpoint.publish(topic, msg.move_data());
  		}
  
  	auto rval = message_count;
  	message_count = 0;
  	return rval;
  	}
  
  size_t Manager::FlushLogBuffers()
  	{
  	DBG_LOG(DBG_BROKER, "Flushing all log buffers");
  	auto rval = 0u;
  
  	for ( auto& lb : log_buffers )
  		rval += lb.Flush(bstate->endpoint, log_batch_size);
  
  	statistics.num_logs_outgoing += rval;
  	return rval;
  	}
  
  void Manager::Error(const char* format, ...)
  	{
  	va_list args;
  	va_start(args, format);
  	auto msg = util::vfmt(format, args);
  	va_end(args);
  
  	if ( script_scope )
  		emit_builtin_error(msg);
  	else
  		reporter->Error("%s", msg);
  	}
  
  bool Manager::AutoPublishEvent(string topic, Val* event)
  	{
  	if ( event->GetType()->Tag() != TYPE_FUNC )
  		{
  		Error("Broker::auto_publish must operate on an event");
  		return false;
  		}
  
  	auto event_val = event->AsFunc();
  	if ( event_val->Flavor() != FUNC_FLAVOR_EVENT )
  		{
  		Error("Broker::auto_publish must operate on an event");
  		return false;
  		}
  
  	auto handler = event_registry->Lookup(event_val->Name());
  	if ( ! handler )
  		{
  		Error("Broker::auto_publish failed to lookup event '%s'", event_val->Name());
  		return false;
  		}
  
  	DBG_LOG(DBG_BROKER, "Enabling auto-publising of event %s to topic %s", handler->Name(),
  	        topic.c_str());
  	handler->AutoPublish(move(topic));
  
  	return true;
  	}
  
  bool Manager::AutoUnpublishEvent(const string& topic, Val* event)
  	{
  	if ( event->GetType()->Tag() != TYPE_FUNC )
  		{
  		Error("Broker::auto_event_stop must operate on an event");
  		return false;
  		}
  
  	auto event_val = event->AsFunc();
  
  	if ( event_val->Flavor() != FUNC_FLAVOR_EVENT )
  		{
  		Error("Broker::auto_event_stop must operate on an event");
  		return false;
  		}
  
  	auto handler = event_registry->Lookup(event_val->Name());
  
  	if ( ! handler )
  		{
  		Error("Broker::auto_event_stop failed to lookup event '%s'", event_val->Name());
  		return false;
  		}
  
  	DBG_LOG(DBG_BROKER, "Disabling auto-publishing of event %s to topic %s", handler->Name(),
  	        topic.c_str());
  	handler->AutoUnpublish(topic);
  
  	return true;
  	}
  
  RecordVal* Manager::MakeEvent(ValPList* args, zeek::detail::Frame* frame)
  	{
  	auto rval = new RecordVal(BifType::Record::Broker::Event);
  	auto arg_vec = make_intrusive<VectorVal>(vector_of_data_type);
  	rval->Assign(1, arg_vec);
  	Func* func = nullptr;
  	scoped_reporter_location srl{frame};
  
  	for ( auto i = 0; i < args->length(); ++i )
  		{
  		auto arg_val = (*args)[i];
  
  		if ( i == 0 )
  			{
  			// Event val must come first.
  
  			if ( arg_val->GetType()->Tag() != TYPE_FUNC )
  				{
  				Error("attempt to convert non-event into an event type");
  				return rval;
  				}
  
  			func = arg_val->AsFunc();
  
  			if ( func->Flavor() != FUNC_FLAVOR_EVENT )
  				{
  				Error("attempt to convert non-event into an event type");
  				return rval;
  				}
  
  			auto num_args = func->GetType()->Params()->NumFields();
  
  			if ( num_args != args->length() - 1 )
  				{
  				Error("bad # of arguments: got %d, expect %d", args->length(), num_args + 1);
  				return rval;
  				}
  
  			rval->Assign(0, func->Name());
  			continue;
  			}
  
  		const auto& got_type = (*args)[i]->GetType();
  		const auto& expected_type = func->GetType()->ParamList()->GetTypes()[i - 1];
  
  		if ( ! same_type(got_type, expected_type) )
  			{
  			rval->Remove(0);
  			Error("event parameter #%d type mismatch, got %s, expect %s", i,
  			      type_name(got_type->Tag()), type_name(expected_type->Tag()));
  			return rval;
  			}
  
  		RecordValPtr data_val;
  
  		if ( same_type(got_type, detail::DataVal::ScriptDataType()) )
  			data_val = {NewRef{}, (*args)[i]->AsRecordVal()};
  		else
  			data_val = detail::make_data_val((*args)[i]);
  
  		if ( ! data_val->HasField(0) )
  			{
  			rval->Remove(0);
  			Error("failed to convert param #%d of type %s to broker data", i,
  			      type_name(got_type->Tag()));
  			return rval;
  			}
  
  		arg_vec->Assign(i - 1, std::move(data_val));
  		}
  
  	return rval;
  	}
  
  bool Manager::Subscribe(const string& topic_prefix)
  	{
  	DBG_LOG(DBG_BROKER, "Subscribing to topic prefix %s", topic_prefix.c_str());
  	bstate->subscriber.add_topic(topic_prefix, ! run_state::detail::zeek_init_done);
  
  	// For backward compatibility, we also may receive messages on
  	// "bro/" topic prefixes in addition to "zeek/".
  	if ( strncmp(topic_prefix.data(), "zeek/", 5) == 0 )
  		{
  		std::string alt_topic = "bro/" + topic_prefix.substr(5);
  		bstate->subscriber.add_topic(std::move(alt_topic), ! run_state::detail::zeek_init_done);
  		}
  
  	return true;
  	}
  
  bool Manager::Forward(string topic_prefix)
  	{
  	for ( const auto& prefix : forwarded_prefixes )
  		if ( prefix == topic_prefix )
  			return false;
  
  	DBG_LOG(DBG_BROKER, "Forwarding topic prefix %s", topic_prefix.c_str());
  	Subscribe(topic_prefix);
  	forwarded_prefixes.emplace_back(std::move(topic_prefix));
  	return true;
  	}
  
  bool Manager::Unsubscribe(const string& topic_prefix)
  	{
  	for ( size_t i = 0; i < forwarded_prefixes.size(); ++i )
  		if ( forwarded_prefixes[i] == topic_prefix )
  			{
  			DBG_LOG(DBG_BROKER, "Unforwading topic prefix %s", topic_prefix.c_str());
  			forwarded_prefixes.erase(forwarded_prefixes.begin() + i);
  			break;
  			}
  
  	DBG_LOG(DBG_BROKER, "Unsubscribing from topic prefix %s", topic_prefix.c_str());
  	bstate->subscriber.remove_topic(topic_prefix, ! run_state::detail::zeek_init_done);
  	return true;
  	}
  
  void Manager::DispatchMessage(const broker::topic& topic, broker::data msg)
  	{
  	switch ( broker::zeek::Message::type(msg) )
  		{
  		case broker::zeek::Message::Type::Invalid:
  			reporter->Warning("received invalid broker message: %s", broker::to_string(msg).data());
  			break;
  
  		case broker::zeek::Message::Type::Event:
  			ProcessEvent(topic, std::move(msg));
  			break;
  
  		case broker::zeek::Message::Type::LogCreate:
  			ProcessLogCreate(std::move(msg));
  			break;
  
  		case broker::zeek::Message::Type::LogWrite:
  			ProcessLogWrite(std::move(msg));
  			break;
  
  		case broker::zeek::Message::Type::IdentifierUpdate:
  			ProcessIdentifierUpdate(std::move(msg));
  			break;
  
  		case broker::zeek::Message::Type::Batch:
  			{
  			broker::zeek::Batch batch(std::move(msg));
  
  			if ( ! batch.valid() )
  				{
  				reporter->Warning("received invalid broker Batch: %s",
  				                  broker::to_string(batch).data());
  				return;
  				}
  
  			for ( auto& i : batch.batch() )
  				DispatchMessage(topic, std::move(i));
  
  			break;
  			}
  
  		default:
  			// We ignore unknown types so that we could add more in the
  			// future if we had too.
  			reporter->Warning("received unknown broker message: %s", broker::to_string(msg).data());
  			break;
  		}
  	}
  
  void Manager::Process()
  	{
  	// Ensure that time gets update before processing broker messages, or events
  	// based on them might get scheduled wrong.
  	if ( use_real_time )
  		run_state::detail::update_network_time(util::current_time());
  
  	auto messages = bstate->subscriber.poll();
  
  	bool had_input = ! messages.empty();
  
  	for ( auto& message : messages )
  		{
  		auto& topic = broker::get_topic(message);
  
  		if ( broker::is_prefix(topic, broker::topic::statuses_str) )
  			{
  			if ( auto stat = broker::make_status_view(get_data(message)) )
  				{
  				ProcessStatus(stat);
  				}
  			else
  				{
  				auto str = to_string(message);
  				reporter->Warning("ignoring malformed Broker status event: %s", str.c_str());
  				}
  			continue;
  			}
  
  		if ( broker::is_prefix(topic, broker::topic::errors_str) )
  			{
  			if ( auto err = broker::make_error_view(get_data(message)) )
  				{
  				ProcessError(err);
  				}
  			else
  				{
  				auto str = to_string(message);
  				reporter->Warning("ignoring malformed Broker error event: %s", str.c_str());
  				}
  			continue;
  			}
  
  		if ( broker::is_prefix(topic, broker::topic::store_events_str) )
  			{
  			ProcessStoreEvent(broker::move_data(message));
  			continue;
  			}
  
  		try
  			{
  			// Once we call a broker::move_* function, we force Broker to
  			// unshare the content of the message, i.e., copy the content to a
  			// different memory region if other threads keep references to the
  			// message. Since `topic` still points into the original memory
  			// region, we may no longer access it after this point.
  			auto unshared_topic = broker::move_topic(message);
  			DispatchMessage(unshared_topic, broker::move_data(message));
  			}
  		catch ( std::runtime_error& e )
  			{
  			reporter->Warning("ignoring invalid Broker message: %s", +e.what());
  			continue;
  			}
  		}
  
  	for ( auto& s : data_stores )
  		{
  		auto num_available = s.second->proxy.mailbox().size();
  
  		if ( num_available > 0 )
  			{
  			had_input = true;
  			auto responses = s.second->proxy.receive(num_available);
  
  			for ( auto& r : responses )
  				ProcessStoreResponse(s.second, move(r));
  			}
  		}
  
  	if ( had_input )
  		{
  		if ( run_state::network_time == 0 )
  			// If we're getting Broker messages, but still haven't initialized
  			// run_state::network_time, may as well do so now because otherwise the
  			// broker/cluster logs will end up using timestamp 0.
  			run_state::detail::update_network_time(util::current_time());
  		}
  	}
  
  void Manager::ProcessStoreEventInsertUpdate(const TableValPtr& table, const std::string& store_id,
                                              const broker::data& key, const broker::data& data,
                                              const broker::data& old_value, bool insert)
  	{
  	auto type = "Insert";
  	if ( ! insert )
  		type = "Update";
  
  	if ( insert )
  		{
  		DBG_LOG(DBG_BROKER, "Store %s: Insert: %s:%s (%s:%s)", store_id.c_str(),
  		        to_string(key).c_str(), to_string(data).c_str(), key.get_type_name(),
  		        data.get_type_name());
  		}
  	else
  		{
  		DBG_LOG(DBG_BROKER, "Store %s: Update: %s->%s (%s)", store_id.c_str(),
  		        to_string(old_value).c_str(), to_string(data).c_str(), data.get_type_name());
  		}
  
  	if ( table->GetType()->IsSet() && data.get_type() != broker::data::type::none )
  		{
  		reporter->Error("ProcessStoreEvent %s got %s when expecting set", type,
  		                data.get_type_name());
  		return;
  		}
  
  	const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
  	ValPtr zeek_key;
  	if ( its.size() == 1 )
  		zeek_key = detail::data_to_val(key, its[0].get());
  	else
  		zeek_key = detail::data_to_val(key, table->GetType()->AsTableType()->GetIndices().get());
  
  	if ( ! zeek_key )
  		{
  		reporter->Error(
  			"ProcessStoreEvent %s: could not convert key \"%s\" for store \"%s\" while receiving "
  			"remote data. This probably means the tables have different types on different nodes.",
  			type, to_string(key).c_str(), store_id.c_str());
  		return;
  		}
  
  	if ( table->GetType()->IsSet() )
  		{
  		table->Assign(zeek_key, nullptr, false);
  		return;
  		}
  
  	// it is a table
  	auto zeek_value = detail::data_to_val(data, table->GetType()->Yield().get());
  	if ( ! zeek_value )
  		{
  		reporter->Error("ProcessStoreEvent %s: could not convert value \"%s\" for key \"%s\" in "
  		                "store \"%s\" while receiving remote data. This probably means the tables "
  		                "have different types on different nodes.",
  		                type, to_string(data).c_str(), to_string(key).c_str(), store_id.c_str());
  		return;
  		}
  
  	table->Assign(zeek_key, zeek_value, false);
  	}
  
  void Manager::ProcessStoreEvent(broker::data msg)
  	{
  	if ( auto insert = broker::store_event::insert::make(msg) )
  		{
  		auto storehandle = broker_mgr->LookupStore(insert.store_id());
  		if ( ! storehandle )
  			return;
  
  		const auto& table = storehandle->forward_to;
  		if ( ! table )
  			return;
  
  		// We sent this message. Ignore it.
  		if ( insert.publisher() == storehandle->store_pid )
  			return;
  
  		ProcessStoreEventInsertUpdate(table, insert.store_id(), insert.key(), insert.value(), {},
  		                              true);
  		}
  	else if ( auto update = broker::store_event::update::make(msg) )
  		{
  		auto storehandle = broker_mgr->LookupStore(update.store_id());
  		if ( ! storehandle )
  			return;
  
  		const auto& table = storehandle->forward_to;
  		if ( ! table )
  			return;
  
  		// We sent this message. Ignore it.
  		if ( update.publisher() == storehandle->store_pid )
  			return;
  
  		ProcessStoreEventInsertUpdate(table, update.store_id(), update.key(), update.new_value(),
  		                              update.old_value(), false);
  		}
  	else if ( auto erase = broker::store_event::erase::make(msg) )
  		{
  		auto storehandle = broker_mgr->LookupStore(erase.store_id());
  		if ( ! storehandle )
  			return;
  
  		auto table = storehandle->forward_to;
  		if ( ! table )
  			return;
  
  		// We sent this message. Ignore it.
  		if ( erase.publisher() == storehandle->store_pid )
  			return;
  
  		auto key = erase.key();
  		DBG_LOG(DBG_BROKER, "Store %s: Erase key %s", erase.store_id().c_str(),
  		        to_string(key).c_str());
  		const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
  		assert(its.size() == 1);
  		auto zeek_key = detail::data_to_val(key, its[0].get());
  		if ( ! zeek_key )
  			{
  			reporter->Error("ProcessStoreEvent: could not convert key \"%s\" for store \"%s\" "
  			                "while receiving remote erase. This probably means the tables have "
  			                "different types on different nodes.",
  			                to_string(key).c_str(), insert.store_id().c_str());
  			return;
  			}
  
  		table->Remove(*zeek_key, false);
  		}
  	else if ( auto expire = broker::store_event::expire::make(msg) )
  		{
  			// We just ignore expiries - expiring information on the Zeek side is handled by Zeek
  			// itself.
  #ifdef DEBUG
  		// let's only debug log for stores that we know.
  		auto storehandle = broker_mgr->LookupStore(expire.store_id());
  		if ( ! storehandle )
  			return;
  
  		auto table = storehandle->forward_to;
  		if ( ! table )
  			return;
  
  		DBG_LOG(DBG_BROKER, "Store %s: Store expired key %s", expire.store_id().c_str(),
  		        to_string(expire.key()).c_str());
  #endif /* DEBUG */
  		}
  	else
  		{
  		reporter->Error("ProcessStoreEvent: Unhandled event type");
  		}
  	}
  
  void Manager::ProcessEvent(const broker::topic& topic, broker::zeek::Event ev)
  	{
  	if ( ! ev.valid() )
  		{
  		reporter->Warning("received invalid broker Event: %s",
  		                  broker::to_string(ev.as_data()).data());
  		return;
  		}
  
  	auto name = std::move(ev.name());
  	auto args = std::move(ev.args());
  
  	DBG_LOG(DBG_BROKER, "Process event: %s %s", name.data(), RenderMessage(args).data());
  	++statistics.num_events_incoming;
  	auto handler = event_registry->Lookup(name);
  
  	if ( ! handler )
  		return;
  
  	auto& topic_string = topic.string();
  
  	for ( const auto& p : forwarded_prefixes )
  		{
  		if ( p.size() > topic_string.size() )
  			continue;
  
  		if ( strncmp(p.data(), topic_string.data(), p.size()) != 0 )
  			continue;
  
  		DBG_LOG(DBG_BROKER, "Skip processing of forwarded event: %s %s", name.data(),
  		        RenderMessage(args).data());
  		return;
  		}
  
  	const auto& arg_types = handler->GetType(false)->ParamList()->GetTypes();
  
  	if ( arg_types.size() != args.size() )
  		{
  		reporter->Warning("got event message '%s' with invalid # of args,"
  		                  " got %zd, expected %zu",
  		                  name.data(), args.size(), arg_types.size());
  		return;
  		}
  
  	Args vl;
  	vl.reserve(args.size());
  
  	for ( size_t i = 0; i < args.size(); ++i )
  		{
  		auto got_type = args[i].get_type_name();
  		const auto& expected_type = arg_types[i];
  		auto val = detail::data_to_val(std::move(args[i]), expected_type.get());
  
  		if ( val )
  			vl.emplace_back(std::move(val));
  		else
  			{
  			auto expected_name = type_name(expected_type->Tag());
  
  			reporter->Warning("failed to convert remote event '%s' arg #%zu,"
  			                  " got %s, expected %s",
  			                  name.data(), i, got_type, expected_name);
  
  			// If we got a vector and expected a function this is
  			// possibly because of a mismatch between
  			// anonymous-function bodies.
  			if ( strcmp(expected_name, "func") == 0 && strcmp("vector", got_type) == 0 )
  				reporter->Warning("when sending functions the receiver must have access to a"
  				                  " version of that function.\nFor anonymous functions, that "
  				                  "function must have the same body.");
  
  			break;
  			}
  		}
  
  	if ( vl.size() == args.size() )
  		event_mgr.Enqueue(handler, std::move(vl), util::detail::SOURCE_BROKER);
  	}
  
  bool Manager::ProcessLogCreate(broker::zeek::LogCreate lc)
  	{
  	DBG_LOG(DBG_BROKER, "Received log-create: %s", RenderMessage(lc.as_data()).c_str());
  	if ( ! lc.valid() )
  		{
  		reporter->Warning("received invalid broker LogCreate: %s", broker::to_string(lc).data());
  		return false;
  		}
  
  	auto stream_id = detail::data_to_val(std::move(lc.stream_id()), log_id_type);
  	if ( ! stream_id )
  		{
  		reporter->Warning("failed to unpack remote log stream id");
  		return false;
  		}
  
  	auto writer_id = detail::data_to_val(std::move(lc.writer_id()), writer_id_type);
  	if ( ! writer_id )
  		{
  		reporter->Warning("failed to unpack remote log writer id");
  		return false;
  		}
  
  	auto writer_info = std::make_unique<logging::WriterBackend::WriterInfo>();
  	if ( ! writer_info->FromBroker(std::move(lc.writer_info())) )
  		{
  		reporter->Warning("failed to unpack remote log writer info");
  		return false;
  		}
  
  	// Get log fields.
  	auto fields_data = caf::get_if<broker::vector>(&lc.fields_data());
  
  	if ( ! fields_data )
  		{
  		reporter->Warning("failed to unpack remote log fields");
  		return false;
  		}
  
  	auto num_fields = fields_data->size();
  	auto fields = new threading::Field*[num_fields];
  
  	for ( size_t i = 0; i < num_fields; ++i )
  		{
  		if ( auto field = detail::data_to_threading_field(std::move((*fields_data)[i])) )
  			fields[i] = field;
  		else
  			{
  			reporter->Warning("failed to convert remote log field # %zu", i);
  			delete[] fields;
  			return false;
  			}
  		}
  
  	if ( ! log_mgr->CreateWriterForRemoteLog(stream_id->AsEnumVal(), writer_id->AsEnumVal(),
  	                                         writer_info.release(), num_fields, fields) )
  		{
  		ODesc d;
  		stream_id->Describe(&d);
  		reporter->Warning("failed to create remote log stream for %s locally", d.Description());
  		}
  
  	return true;
  	}
  
  bool Manager::ProcessLogWrite(broker::zeek::LogWrite lw)
  	{
  	DBG_LOG(DBG_BROKER, "Received log-write: %s", RenderMessage(lw.as_data()).c_str());
  
  	if ( ! lw.valid() )
  		{
  		reporter->Warning("received invalid broker LogWrite: %s", broker::to_string(lw).data());
  		return false;
  		}
  
  	++statistics.num_logs_incoming;
  	auto& stream_id_name = lw.stream_id().name;
  
  	// Get stream ID.
  	auto stream_id = detail::data_to_val(std::move(lw.stream_id()), log_id_type);
  
  	if ( ! stream_id )
  		{
  		reporter->Warning("failed to unpack remote log stream id: %s", stream_id_name.data());
  		return false;
  		}
  
  	// Get writer ID.
  	auto writer_id = detail::data_to_val(std::move(lw.writer_id()), writer_id_type);
  	if ( ! writer_id )
  		{
  		reporter->Warning("failed to unpack remote log writer id for stream: %s",
  		                  stream_id_name.data());
  		return false;
  		}
  
  	auto path = caf::get_if<std::string>(&lw.path());
  
  	if ( ! path )
  		{
  		reporter->Warning("failed to unpack remote log values (bad path variant) for stream: %s",
  		                  stream_id_name.data());
  		return false;
  		}
  
  	auto serial_data = caf::get_if<std::string>(&lw.serial_data());
  
  	if ( ! serial_data )
  		{
  		reporter->Warning(
  			"failed to unpack remote log values (bad serial_data variant) for stream: %s",
  			stream_id_name.data());
  		return false;
  		}
  
  	zeek::detail::BinarySerializationFormat fmt;
  	fmt.StartRead(serial_data->data(), serial_data->size());
  
  	int num_fields;
  	bool success = fmt.Read(&num_fields, "num_fields");
  
  	if ( ! success )
  		{
  		reporter->Warning("failed to unserialize remote log num fields for stream: %s",
  		                  stream_id_name.data());
  		return false;
  		}
  
  	auto vals = new threading::Value*[num_fields];
  
  	for ( int i = 0; i < num_fields; ++i )
  		{
  		vals[i] = new threading::Value;
  
  		if ( ! vals[i]->Read(&fmt) )
  			{
  			for ( int j = 0; j <= i; ++j )
  				delete vals[j];
  
  			delete[] vals;
  			reporter->Warning("failed to unserialize remote log field %d for stream: %s", i,
  			                  stream_id_name.data());
  
  			return false;
  			}
  		}
  
  	log_mgr->WriteFromRemote(stream_id->AsEnumVal(), writer_id->AsEnumVal(), std::move(*path),
  	                         num_fields, vals);
  	fmt.EndRead();
  	return true;
  	}
  
  bool Manager::ProcessIdentifierUpdate(broker::zeek::IdentifierUpdate iu)
  	{
  	DBG_LOG(DBG_BROKER, "Received id-update: %s", RenderMessage(iu.as_data()).c_str());
  
  	if ( ! iu.valid() )
  		{
  		reporter->Warning("received invalid broker IdentifierUpdate: %s",
  		                  broker::to_string(iu).data());
  		return false;
  		}
  
  	++statistics.num_ids_incoming;
  	auto id_name = std::move(iu.id_name());
  	auto id_value = std::move(iu.id_value());
  	const auto& id = zeek::detail::global_scope()->Find(id_name);
  
  	if ( ! id )
  		{
  		reporter->Warning("Received id-update request for unkown id: %s", id_name.c_str());
  		return false;
  		}
  
  	auto val = detail::data_to_val(std::move(id_value), id->GetType().get());
  
  	if ( ! val )
  		{
  		reporter->Error("Failed to receive ID with unsupported type: %s (%s)", id_name.c_str(),
  		                type_name(id->GetType()->Tag()));
  		return false;
  		}
  
  	id->SetVal(std::move(val));
  	return true;
  	}
  
  void Manager::ProcessStatus(broker::status_view stat)
  	{
  	DBG_LOG(DBG_BROKER, "Received status message: %s", RenderMessage(stat).c_str());
  
  	auto ctx = stat.context();
  
  	EventHandlerPtr event;
  	switch ( stat.code() )
  		{
  		case broker::sc::unspecified:
  			event = ::Broker::status;
  			break;
  
  		case broker::sc::peer_added:
  			++peer_count;
  			assert(ctx);
  			log_mgr->SendAllWritersTo(*ctx);
  			event = ::Broker::peer_added;
  			break;
  
  		case broker::sc::peer_removed:
  			--peer_count;
  			event = ::Broker::peer_removed;
  			break;
  
  		case broker::sc::peer_lost:
  			--peer_count;
  			event = ::Broker::peer_lost;
  			break;
  
  		default:
  			reporter->Warning("Unhandled Broker status: %s", to_string(stat).data());
  			break;
  		}
  
  	if ( ! event )
  		return;
  
  	static auto ei = id::find_type<RecordType>("Broker::EndpointInfo");
  	auto endpoint_info = make_intrusive<RecordVal>(ei);
  
  	if ( ctx )
  		{
  		endpoint_info->Assign(0, to_string(ctx->node));
  		static auto ni = id::find_type<RecordType>("Broker::NetworkInfo");
  		auto network_info = make_intrusive<RecordVal>(ni);
  
  		if ( ctx->network )
  			{
  			network_info->Assign(0, ctx->network->address.data());
  			network_info->Assign(1, val_mgr->Port(ctx->network->port, TRANSPORT_TCP));
  			}
  		else
  			{
  			// TODO: are there any status messages where the ctx->network
  			// is not set and actually could be?
  			network_info->Assign(0, "<unknown>");
  			network_info->Assign(1, val_mgr->Port(0, TRANSPORT_TCP));
  			}
  
  		endpoint_info->Assign(1, std::move(network_info));
  		}
  
  	auto str = stat.message();
  	auto msg = make_intrusive<StringVal>(str ? *str : "");
  
  	event_mgr.Enqueue(event, std::move(endpoint_info), std::move(msg));
  	}
  
  void Manager::ProcessError(broker::error_view err)
  	{
  	DBG_LOG(DBG_BROKER, "Received error message: %s", RenderMessage(err).c_str());
  
  	if ( ! ::Broker::error )
  		return;
  
  	auto int_code = static_cast<uint8_t>(err.code());
  
  	BifEnum::Broker::ErrorCode ec;
  	static auto enum_type = id::find_type<EnumType>("Broker::ErrorCode");
  	if ( enum_type->Lookup(int_code) )
  		ec = static_cast<BifEnum::Broker::ErrorCode>(int_code);
  	else
  		{
  
  		reporter->Warning("Unknown Broker error code %u: mapped to unspecificed enum value ",
  		                  static_cast<unsigned>(int_code));
  		ec = BifEnum::Broker::ErrorCode::UNSPECIFIED;
  		}
  
  	std::string msg;
  	// Note: we could also use to_string, but that would change the log output
  	// and we would have to update all baselines relying on this format.
  	if ( auto ctx = err.context() )
  		{
  		msg += '(';
  		msg += to_string(ctx->node);
  		msg += ", ";
  		msg += caf::deep_to_string(ctx->network);
  		msg += ", ";
  		if ( auto what = err.message() )
  			msg += caf::deep_to_string(*what);
  		else
  			msg += R"_("")_";
  		msg += ')';
  		}
  	else
  		msg = "(null)";
  
  	event_mgr.Enqueue(::Broker::error, BifType::Enum::Broker::ErrorCode->GetEnumVal(ec),
  	                  make_intrusive<StringVal>(msg));
  	}
  
  void Manager::ProcessStoreResponse(detail::StoreHandleVal* s, broker::store::response response)
  	{
  	DBG_LOG(DBG_BROKER, "Received store response: %s", RenderMessage(response).c_str());
  
  	auto request = pending_queries.find(std::make_pair(response.id, s));
  
  	if ( request == pending_queries.end() )
  		{
  		reporter->Warning("unmatched response to query %" PRIu64 " on store %s", response.id,
  		                  s->store.name().c_str());
  		return;
  		}
  
  	if ( request->second->Disabled() )
  		{
  		// Trigger timer must have timed the query out already.
  		delete request->second;
  		pending_queries.erase(request);
  		return;
  		}
  
  	if ( response.answer )
  		request->second->Result(
  			detail::query_result(detail::make_data_val(std::move(*response.answer))));
  	else if ( response.answer.error() == broker::ec::request_timeout )
  		{
  		// Fine, trigger's timeout takes care of things.
  		}
  	else if ( response.answer.error() == broker::ec::stale_data )
  		{
  		// It's sort of arbitrary whether to make this type of error successful
  		// query with a "fail" status versus going through the when stmt timeout
  		// code path.  I think the timeout path is maybe more expected in order
  		// for failures like "no such key" to actually be distinguishable from
  		// this type of error (which is less easily handled programmatically).
  		}
  	else if ( response.answer.error() == broker::ec::no_such_key )
  		request->second->Result(detail::query_result());
  	else
  		reporter->InternalWarning("unknown store response status: %s",
  		                          to_string(response.answer.error()).c_str());
  
  	delete request->second;
  	pending_queries.erase(request);
  	}
  
  detail::StoreHandleVal* Manager::MakeMaster(const string& name, broker::backend type,
                                              broker::backend_options opts)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return nullptr;
  
  	if ( LookupStore(name) )
  		return nullptr;
  
  	DBG_LOG(DBG_BROKER, "Creating master for data store %s", name.c_str());
  
  	auto it = opts.find("path");
  
  	if ( it == opts.end() )
  		it = opts.emplace("path", "").first;
  
  	if ( it->second == broker::data("") )
  		{
  		auto suffix = ".store";
  
  		switch ( type )
  			{
  			case broker::backend::sqlite:
  				suffix = ".sqlite";
  				break;
  			default:
  				break;
  			}
  
  		it->second = name + suffix;
  		}
  
  	auto result = bstate->endpoint.attach_master(name, type, move(opts));
  	if ( ! result )
  		{
  		Error("Failed to attach master store %s:", to_string(result.error()).c_str());
  		return nullptr;
  		}
  
  	auto handle = new detail::StoreHandleVal{*result};
  	Ref(handle);
  
  	data_stores.emplace(name, handle);
  	iosource_mgr->RegisterFd(handle->proxy.mailbox().descriptor(), this);
  	PrepareForwarding(name);
  
  	if ( ! bstate->endpoint.use_real_time() )
  		// Wait for master to become available/responsive.
  		// Possibly avoids timeouts in scripts during unit tests.
  		handle->store.exists("");
  
  	BrokerStoreToZeekTable(name, handle);
  
  	return handle;
  	}
  
  void Manager::BrokerStoreToZeekTable(const std::string& name, const detail::StoreHandleVal* handle)
  	{
  	if ( ! handle->forward_to )
  		return;
  
  	auto keys = handle->store.keys();
  	if ( ! keys )
  		return;
  
  	auto set = caf::get_if<broker::set>(&(keys->get_data()));
  	auto table = handle->forward_to;
  	const auto& its = table->GetType()->AsTableType()->GetIndexTypes();
  	bool is_set = table->GetType()->IsSet();
  
  	// disable &on_change notifications while filling the table.
  	table->DisableChangeNotifications();
  
  	for ( const auto& key : *set )
  		{
  		ValPtr zeek_key;
  		if ( its.size() == 1 )
  			zeek_key = detail::data_to_val(key, its[0].get());
  		else
  			zeek_key = detail::data_to_val(key,
  			                               table->GetType()->AsTableType()->GetIndices().get());
  
  		if ( ! zeek_key )
  			{
  			reporter->Error("Failed to convert key \"%s\" while importing broker store to table "
  			                "for store \"%s\". Aborting import.",
  			                to_string(key).c_str(), name.c_str());
  			// just abort - this probably means the types are incompatible
  			table->EnableChangeNotifications();
  			return;
  			}
  
  		if ( is_set )
  			{
  			table->Assign(zeek_key, nullptr, false);
  			continue;
  			}
  
  		auto value = handle->store.get(key);
  		if ( ! value )
  			{
  			reporter->Error(
  				"Failed to load value for key %s while importing Broker store %s to table",
  				to_string(key).c_str(), name.c_str());
  			table->EnableChangeNotifications();
  			continue;
  			}
  
  		auto zeek_value = detail::data_to_val(*value, table->GetType()->Yield().get());
  		if ( ! zeek_value )
  			{
  			reporter->Error("Could not convert %s to table value while trying to import Broker "
  			                "store %s. Aborting import.",
  			                to_string(value).c_str(), name.c_str());
  			table->EnableChangeNotifications();
  			return;
  			}
  
  		table->Assign(zeek_key, zeek_value, false);
  		}
  
  	table->EnableChangeNotifications();
  	return;
  	}
  
  detail::StoreHandleVal* Manager::MakeClone(const string& name, double resync_interval,
                                             double stale_interval, double mutation_buffer_interval)
  	{
  	if ( bstate->endpoint.is_shutdown() )
  		return nullptr;
  
  	if ( LookupStore(name) )
  		return nullptr;
  
  	DBG_LOG(DBG_BROKER, "Creating clone for data store %s", name.c_str());
  
  	auto result = bstate->endpoint.attach_clone(name, resync_interval, stale_interval,
  	                                            mutation_buffer_interval);
  	if ( ! result )
  		{
  		Error("Failed to attach clone store %s:", to_string(result.error()).c_str());
  		return nullptr;
  		}
  
  	auto handle = new detail::StoreHandleVal{*result};
  	Ref(handle);
  
  	data_stores.emplace(name, handle);
  	iosource_mgr->RegisterFd(handle->proxy.mailbox().descriptor(), this);
  	PrepareForwarding(name);
  	return handle;
  	}
  
  detail::StoreHandleVal* Manager::LookupStore(const string& name)
  	{
  	auto i = data_stores.find(name);
  	return i == data_stores.end() ? nullptr : i->second;
  	}
  
  bool Manager::CloseStore(const string& name)
  	{
  	DBG_LOG(DBG_BROKER, "Closing data store %s", name.c_str());
  
  	auto s = data_stores.find(name);
  	if ( s == data_stores.end() )
  		return false;
  
  	iosource_mgr->UnregisterFd(s->second->proxy.mailbox().descriptor(), this);
  
  	for ( auto i = pending_queries.begin(); i != pending_queries.end(); )
  		if ( i->second->Store().name() == name )
  			{
  			i->second->Abort();
  			delete i->second;
  			i = pending_queries.erase(i);
  			}
  		else
  			{
  			++i;
  			}
  
  	s->second->have_store = false;
  	s->second->store_pid = {};
  	s->second->proxy = {};
  	s->second->store = {};
  	Unref(s->second);
  	data_stores.erase(s);
  	return true;
  	}
  
  bool Manager::TrackStoreQuery(detail::StoreHandleVal* handle, broker::request_id id,
                                detail::StoreQueryCallback* cb)
  	{
  	auto rval = pending_queries.emplace(std::make_pair(id, handle), cb).second;
  
  	if ( bstate->endpoint.use_real_time() )
  		return rval;
  
  	FlushPendingQueries();
  	return rval;
  	}
  
  const Stats& Manager::GetStatistics()
  	{
  	statistics.num_peers = peer_count;
  	statistics.num_stores = data_stores.size();
  	statistics.num_pending_queries = pending_queries.size();
  
  	// The other attributes are set as activity happens.
  
  	return statistics;
  	}
  
  bool Manager::AddForwardedStore(const std::string& name, TableValPtr table)
  	{
  	if ( forwarded_stores.find(name) != forwarded_stores.end() )
  		{
  		reporter->Error("same &broker_store %s specified for two different variables",
  		                name.c_str());
  		return false;
  		}
  
  	DBG_LOG(DBG_BROKER, "Adding table forward for data store %s", name.c_str());
  	forwarded_stores.emplace(name, std::move(table));
  
  	PrepareForwarding(name);
  	return true;
  	}
  
  void Manager::PrepareForwarding(const std::string& name)
  	{
  	auto handle = LookupStore(name);
  	if ( ! handle )
  		return;
  
  	if ( forwarded_stores.find(name) == forwarded_stores.end() )
  		return;
  
  	handle->forward_to = forwarded_stores.at(name);
  	DBG_LOG(DBG_BROKER, "Resolved table forward for data store %s", name.c_str());
  	}
  
  void Manager::SetMetricsExportInterval(double value)
  	{
  	broker::timespan ts;
  	if ( broker::convert(value, ts) )
  		bstate->endpoint.metrics_exporter().set_interval(ts);
  	}
  
  void Manager::SetMetricsExportTopic(std::string value)
  	{
  	bstate->endpoint.metrics_exporter().set_target(std::move(value));
  	}
  
  void Manager::SetMetricsExportEndpointName(std::string value)
  	{
  	bstate->endpoint.metrics_exporter().set_id(std::move(value));
  	}
  
  void Manager::SetMetricsExportPrefixes(std::vector<std::string> filter)
  	{
  	bstate->endpoint.metrics_exporter().set_prefixes(std::move(filter));
  	}
  
  std::unique_ptr<telemetry::Manager> Manager::NewTelemetryManager()
  	{
  	// The telemetry Manager actually only has a dependency on the actor system,
  	// not to the Broker Manager. By having the telemetry Manager hold on to a
  	// shared_ptr to our Broker state, we make sure the Broker endpoint, which
  	// owns the CAF actor system, lives for as long as necessary. This also
  	// makes sure that the Broker Manager may even get destroyed before the
  	// telemetry Manager.
  	struct TM final : public telemetry::Manager
  		{
  		using MetricRegistryPtr = std::unique_ptr<caf::telemetry::metric_registry>;
  
  		static auto getPimpl(BrokerState& st)
  			{
  			auto registry = std::addressof(st.endpoint.system().metrics());
  			return reinterpret_cast<telemetry::Manager::Impl*>(registry);
  			}
  
  		static auto getPimpl(MetricRegistryPtr& ptr)
  			{
  			return reinterpret_cast<telemetry::Manager::Impl*>(ptr.get());
  			}
  
  		explicit TM(Broker::Manager* parent, MetricRegistryPtr ptr)
  			: telemetry::Manager(getPimpl(ptr)), parent(parent), tmp(std::move(ptr))
  			{
  			assert(tmp != nullptr);
  			assert(parent != nullptr);
  			}
  
  		void InitPostScript() override
  			{
  			assert(parent->bstate != nullptr);
  			ptr = parent->bstate;
  			auto registry = std::addressof(ptr->endpoint.system().metrics());
  			registry->merge(*tmp);
  			tmp.reset();
  			pimpl = reinterpret_cast<telemetry::Manager::Impl*>(registry);
  			}
  
  		Broker::Manager* parent;
  		MetricRegistryPtr tmp;
  		std::shared_ptr<BrokerState> ptr;
  		};
  
  	auto tmp = std::make_unique<caf::telemetry::metric_registry>();
  	return std::make_unique<TM>(this, std::move(tmp));
  	}
  
  	} // namespace zeek::Broker
  